QuantMiroFish/ensemble.py at main · workimdev/QuantMiroFish · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
"""애널리스트 신호 앙상블 - 가중 결합으로 최종 매매 판단"""

import pandas as pd
from analysts.base_analyst import BaseAnalyst, AnalystSignal
import config


def combine_signals(
    analyst_signals: dict[str, AnalystSignal],
    weights: dict[str, float],
) -> dict:
    """
    가중 평균으로 애널리스트 신호 결합.
    weighted_signal = signal × confidence × weight
    """
    weighted_sum = 0.0
    weight_total = 0.0
    confidence_sum = 0.0

    for name, sig in analyst_signals.items():
        w = weights.get(name, 0.0)
        weighted_sum += sig["signal"] * sig["confidence"] * w
        weight_total += sig["confidence"] * w
        confidence_sum += sig["confidence"] * w

    if weight_total > 0:
        final_signal = weighted_sum / weight_total
        final_confidence = confidence_sum / sum(weights.values())
    else:
        final_signal = 0.0
        final_confidence = 0.0

    # 임계값 기반 포지션 결정
    if final_signal > config.SIGNAL_THRESHOLD:
        position = "BUY"
    elif final_signal < -config.SIGNAL_THRESHOLD:
        position = "SELL"
    else:
        position = "HOLD"

    return {
        "final_signal": round(final_signal, 4),
        "final_confidence": round(final_confidence, 4),
        "position": position,
        "analyst_signals": analyst_signals,
    }


def run_ensemble(
    df: pd.DataFrame,
    analysts: list[BaseAnalyst],
    weights: dict[str, float] = None,
    progress_callback=None,
) -> pd.DataFrame:
    """
    전체 데이터에 대해 모든 애널리스트 실행 후 앙상블 결합.
    Returns: 시그널 DataFrame (Signal, Confidence, Position + 개별 애널리스트 컬럼)
    """
    if weights is None:
        weights = config.ANALYST_WEIGHTS

    print(f"[앙상블] {len(analysts)}명의 애널리스트로 {len(df)}일 분석 중...")

    records = []
    total = len(df)
    report_interval = max(1, total // 10)

    for idx in range(total):
        if idx % report_interval == 0:
            pct = idx / total * 100
            print(f"  진행률: {pct:.0f}% ({idx}/{total})")
            if progress_callback:
                progress_callback(idx, total, analyst_signals if idx > 0 else None)

        # 각 애널리스트 분석
        analyst_signals = {}
        for analyst in analysts:
            sig = analyst.analyze(df, idx)
            analyst_signals[analyst.name] = sig

        # 앙상블 결합
        result = combine_signals(analyst_signals, weights)

        record = {
            "Date": df.index[idx],
            "Signal": result["final_signal"],
            "Confidence": result["final_confidence"],
            "Position": result["position"],
        }
        # 개별 애널리스트 신호
        for aname, sig in analyst_signals.items():
            record[f"{aname}_signal"] = sig["signal"]
            record[f"{aname}_confidence"] = sig["confidence"]

        records.append(record)

    signals_df = pd.DataFrame(records).set_index("Date")
    print(f"[앙상블] 분석 완료: BUY {(signals_df['Position']=='BUY').sum()}일, "
          f"SELL {(signals_df['Position']=='SELL').sum()}일, "
          f"HOLD {(signals_df['Position']=='HOLD').sum()}일")
    return signals_df