트레이딩 시스템에 머신러닝을 적용한 기술 여정

# 기존 방식: 하드코딩된 규칙
def should_signal(pattern):
    if (pattern.uptrend_gain > 3.0 and
        pattern.volume_ratio < 0.5 and
        pattern.breakout_volume > threshold):
        return True  # 신호 생성
    return False

# ML 방식: 데이터 기반 학습
def evaluate_signal(pattern_features) -> (bool, float):
    """
    26개의 패턴 특성을 입력받아 신호 유효성 예측

    Returns:
        (신호 허용 여부, 예측 정확도)
    """
    ml_prob = model.predict(pattern_features)  # 0.0 ~ 1.0
    return ml_prob >= 0.5, ml_prob  # 50% 이상이면 허용

# 속도 비교 (1000회 추론)
LightGBM:      0.12초  ✅
XGBoost:       0.18초
Random Forest: 1.20초  ❌ (실시간 사용 불가)
LSTM:          0.50초

# 어떤 특성이 중요한지 분석 가능
feature_importance = model.feature_importance()

결과:
1. volume_ratio_support_to_uptrend: 0.18  ← 가장 중요!
2. breakout_body: 0.15
3. hour: 0.12  ← 시간대도 중요
4. uptrend_gain: 0.10

# Early Stopping으로 과적합 방지
model = lgb.train(
    params,
    train_set,
    num_boost_round=1000,
    valid_sets=[valid_set],
    callbacks=[lgb.early_stopping(50)]  # 50라운드 개선 없으면 중단
)

# 1. 실시간 패턴 로깅
class PatternDataLogger:
    """시스템 동작 중 발생한 패턴을 DB에 자동 저장"""

    def log_pattern(self, code, signal_strength):
        pattern_data = {
            'stage_1': {...},    # 1단계 특성
            'stage_2': {...},    # 2단계 특성
            'stage_3': {...},    # 3단계 특성
            'stage_4': {...}     # 4단계 특성
        }
        self.db.insert(pattern_data)

# 2. 일일 결과 데이터 생성 (매일 자동 실행)

1. DB에서 당일 패턴 추출
2. 시뮬레이션으로 성공/실패 여부 계산
3. CSV 파일로 저장

결과 예시:
code, time, feature_1, feature_2, ..., label
XXXX, 09:30, 3.52, 1.23, ..., 1 (성공)
YYYY, 10:15, 2.87, 0.95, ..., 0 (실패)

features = {
    'uptrend_candles': 7,           # 상승 봉 개수
    'uptrend_gain': 3.52,           # 상승률 (%)
    'uptrend_max_volume': 250000,   # 최대 거래량
    'uptrend_avg_body': 0.85,       # 평균 몸통 크기 (%)
    'uptrend_total_volume': 1500000 # 총 거래량
}

features = {
    # 하락 시 거래량이 상승 대비 얼마나 줄었는가?
    'volume_ratio_decline_to_uptrend': 0.48,  # ← 높으면 위험

    # 지지 시 거래량이 상승 대비 얼마나 줄었는가?
    'volume_ratio_support_to_uptrend': 0.32,  # ← 낮을수록 좋음!

    # 돌파 시 거래량이 상승 대비 얼마나 증가했는가?
    'volume_ratio_breakout_to_uptrend': 0.65,  # ← 높을수록 좋음

    # 상승 대비 하락이 얼마나 완만했는가?
    'price_gain_to_decline_ratio': 2.86,

    # 지지 기간이 하락 대비 얼마나 길었는가?
    'candle_ratio_support_to_decline': 1.25
}

features = {
    'hour': 9,                      # 시간
    'minute': 30,                   # 분
    'time_in_minutes': 570,         # 분 단위 시간
    'is_morning': 1,                # 오전 여부
    'signal_type': 1,               # STRONG_BUY=1
    'confidence': 85                # 신뢰도
}

# 발견된 차이점:
실시간: uptrend_gain = 3.52  (percentage)
시뮬:   uptrend_gain = 352000 (원 단위)

# → 단위 불일치!

def extract_features_from_pattern(self, pattern: Dict) -> pd.DataFrame:
    """
    시뮬레이션과 실시간 거래 양쪽 데이터 구조 지원

    지원 구조:
    1. pattern_stages: {1_uptrend, 2_decline, 3_support, 4_breakout}
    2. debug_info: {uptrend, decline, support, breakout}
    """

    # 🔄 두 구조 모두 지원
    pattern_stages = pattern.get('pattern_stages', {})
    debug_info = pattern.get('debug_info', {})

    # 상승 구간: 우선순위로 탐색
    uptrend = pattern_stages.get('1_uptrend', debug_info.get('uptrend', {}))

    # 필드명 통일 (gain_pct vs price_gain)
    uptrend_gain = self._safe_float(
        uptrend.get('gain_pct', uptrend.get('price_gain', 0.0))
    )

    # 안전한 형변환
    def _safe_float(self, value, default=0.0):
        if isinstance(value, str):
            # "3.52%" → 3.52, "162,154" → 162154
            value = value.replace(',', '').replace('%', '')
        return float(value) if value else default

# 1. 실시간 시스템 동작 중: 메모리 데이터 덤프
# memory_data_YYYYMMDD_HHMMSS.txt

# 2. 이후: 덤프 데이터로 시뮬 재실행
python replay_from_memory.py memory_data_YYYYMMDD_HHMMSS.txt

# 3. 결과 비교
실시간 시스템: 09:30 → ML 39.9% (차단)
리플레이:      09:30 → ML 39.9% (차단) ✅ 완전 일치!

# 모델 저장 시 메타데이터 포함
model_data = {
    'model': lgb_model,
    'label_encoder': le,
    'feature_names': feature_names,
    'version': f'v_{datetime.now().strftime("%Y%m%d_%H%M%S")}',
    'train_score': 0.756,
    'test_score': 0.729,
    'train_date': datetime.now().strftime("%Y-%m-%d"),
    'data_days': 60
}

with open('model.pkl', 'wb') as f:
    pickle.dump(model_data, f)

# 매 실행 후 정확도 계산
class MLPerformanceTracker:
    def __init__(self):
        self.predictions = []
        self.actuals = []

    def log_result(self, ml_prob, success):
        self.predictions.append(ml_prob)
        self.actuals.append(1 if success else 0)

    def get_weekly_stats(self):
        """주간 성능 리포트"""
        return {
            'accuracy': accuracy_score(self.actuals,
                                      [1 if p >= 0.5 else 0 for p in self.predictions]),
            'avg_prob_success': np.mean([p for p, a in zip(self.predictions, self.actuals) if a == 1]),
            'avg_prob_fail': np.mean([p for p, a in zip(self.predictions, self.actuals) if a == 0])
        }

# 성능 저하 시 텔레그램 알림
if weekly_accuracy < 0.60:
    send_telegram_alert(
        f"⚠️ ML 모델 성능 저하 감지!\n"
        f"주간 정확도: {weekly_accuracy:.1%}\n"
        f"재학습 권장"
    )

# LightGBM + XGBoost + CatBoost 앙상블
lgb_prob = lgb_model.predict(features)
xgb_prob = xgb_model.predict(features)
cat_prob = cat_model.predict(features)

final_prob = (lgb_prob * 0.5 + xgb_prob * 0.3 + cat_prob * 0.2)

# 실행 결과를 즉시 학습 데이터에 추가
from river import linear_model

model = linear_model.LogisticRegression()

# 스트리밍 학습
for features, result in system_results:
    prediction = model.predict_one(features)
    model.learn_one(features, result)  # 즉시 업데이트!