예제 소스

compare_lora_heuristic.py

"""LoRA 모의매매 결과를 동일 구간의 휴리스틱·단순보유와 비교한다.

CSV에 시각 열이 없으므로 ``--bars-per-day``는 가정값이다. 기본값 288은
현재 수집 주기인 5분봉을 가정하며, 결과에 이 가정을 명시한다. 모든 휴리스틱은 현재와 과거
행만 사용한다. 미래 가격은 폭락·폭등 신호 사후 진단에만 사용한다.
"""

from __future__ import annotations

import argparse
import json
import math
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable

import numpy as np
import pandas as pd

from evaluate_qwen_chart_lora import (
    BUY_BALANCE_RATIO,
    DEFAULT_INITIAL_KRW,
    FEE,
    PaperAccount,
)

ACTION_NAMES = {0: "매도", 1: "매수", 2: "유지"}


@dataclass
class StrategyResult:
    name: str
    initial_krw: float
    final_liquidation_value: float
    liquidation_return_pct: float
    max_drawdown_pct: float
    completed_trades: int
    profitable_trades: int
    win_rate_pct: float
    buy_count: int
    sell_count: int
    open_position: bool


def _as_bool(series: pd.Series) -> pd.Series:
    return series.astype(str).str.lower().isin(("1", "1.0", "true", "yes"))


def _simulate(
    prices: Iterable[float],
    actions: Iterable[int],
    *,
    name: str,
    initial_krw: float,
) -> StrategyResult:
    account = PaperAccount.create(initial_krw)
    peak = float(initial_krw)
    max_drawdown = 0.0
    last_price = 0.0
    for price, action in zip(prices, actions):
        last_price = float(price)
        account.execute(int(action), last_price)
        value = account.liquidation_value(last_price)
        peak = max(peak, value)
        max_drawdown = max(max_drawdown, (peak - value) / peak * 100.0)
    final_value = account.liquidation_value(last_price)
    return StrategyResult(
        name=name,
        initial_krw=float(initial_krw),
        final_liquidation_value=final_value,
        liquidation_return_pct=(final_value / initial_krw - 1.0) * 100.0,
        max_drawdown_pct=max_drawdown,
        completed_trades=account.completed_trades,
        profitable_trades=account.profitable_trades,
        win_rate_pct=account.profitable_trades / max(account.completed_trades, 1) * 100.0,
        buy_count=account.buy_count,
        sell_count=account.sell_count,
        open_position=account.unit > 0,
    )


def buy_hold_actions(length: int) -> np.ndarray:
    actions = np.full(int(length), 2, dtype=np.int64)
    if length:
        actions[0] = 1
    return actions


def minmax_actions(
    frame: pd.DataFrame,
    *,
    window: int = 50,
    buy_threshold: float = 10.0,
    sell_threshold: float = 80.0,
) -> np.ndarray:
    """과거 봉만으로 구간 위치를 계산하는 기존 10/80 방식."""
    close = pd.to_numeric(frame["close"], errors="coerce").astype(float)
    low = close.rolling(int(window), min_periods=int(window)).min()
    high = close.rolling(int(window), min_periods=int(window)).max()
    smooth = close.rolling(5, min_periods=5).mean()
    span = (high - low).replace(0.0, np.nan)
    position = ((smooth - low) / span * 100.0).fillna(50.0)
    actions = np.full(len(frame), 2, dtype=np.int64)
    in_position = False
    for i, value in enumerate(position.to_numpy(dtype=float)):
        if not in_position and value <= float(buy_threshold):
            actions[i] = 1
            in_position = True
        elif in_position and value >= float(sell_threshold):
            actions[i] = 0
            in_position = False
    return actions


def trend_heuristic_actions(frame: pd.DataFrame) -> np.ndarray:
    """하락 칼날을 피하고 확인된 반전 뒤에만 진입하는 순행 휴리스틱."""
    close = pd.to_numeric(frame["close"], errors="coerce").astype(float)
    ma20 = close.rolling(20, min_periods=5).mean()
    ma60 = close.rolling(60, min_periods=10).mean()
    ma20_slope = ma20.pct_change(3)
    macd = pd.to_numeric(frame.get("macd", pd.Series(0.0, index=frame.index)), errors="coerce").fillna(0.0)
    macd_slope = macd.diff(3)
    rsi = pd.to_numeric(frame.get("rsi", pd.Series(50.0, index=frame.index)), errors="coerce").fillna(50.0)
    ret20 = close.pct_change(20)
    crash = _as_bool(frame.get("ext_crash_signal", pd.Series(False, index=frame.index)))

    actions = np.full(len(frame), 2, dtype=np.int64)
    in_position = False
    entry = 0.0
    peak = 0.0
    for i in range(len(frame)):
        price = float(close.iloc[i])
        falling = bool(
            i >= 20
            and price < float(ma20.iloc[i])
            and float(ma20_slope.iloc[i]) < 0
            and float(ret20.iloc[i]) < 0
        )
        recovery = bool(
            i >= 60
            and price > float(ma20.iloc[i])
            and float(ma20.iloc[i]) > float(ma60.iloc[i])
            and float(ma20_slope.iloc[i]) > 0
            and float(macd_slope.iloc[i]) > 0
            and 42.0 <= float(rsi.iloc[i]) <= 74.0
        )
        if not in_position:
            if recovery and not bool(crash.iloc[i]):
                actions[i] = 1
                in_position = True
                entry = peak = price
            continue

        peak = max(peak, price)
        net_return = price / entry * ((1.0 - FEE) ** 2) - 1.0
        trailing_return = price / peak - 1.0
        reversal = bool(
            i >= 20
            and price < float(ma20.iloc[i])
            and float(ma20_slope.iloc[i]) < 0
            and float(macd_slope.iloc[i]) < 0
        )
        should_sell = (
            (net_return > 0.002 and reversal)
            or (net_return > 0.003 and trailing_return <= -0.012)
            or (net_return <= -0.03 and falling)
            or (bool(crash.iloc[i]) and falling)
        )
        if should_sell:
            actions[i] = 0
            in_position = False
            entry = peak = 0.0
    return actions


def rl_policy_actions(
    frame: pd.DataFrame,
    checkpoint: Path,
    *,
    initial_krw: float,
) -> np.ndarray:
    """저장된 17특성 REINFORCE 정책을 결정론적으로 재생한다."""
    import torch

    from lib.rl_chart_trainer import RlChartPolicy, build_state_vector

    policy = RlChartPolicy()
    state = torch.load(checkpoint, map_location="cpu", weights_only=True)
    policy.load_state_dict(state)
    policy.eval()
    account = PaperAccount.create(initial_krw)
    actions = np.full(len(frame), 2, dtype=np.int64)
    with torch.inference_mode():
        for i in range(len(frame)):
            price = float(frame.iloc[i]["close"])
            vector = build_state_vector(frame, i, account, price)
            logits = policy(torch.from_numpy(vector).unsqueeze(0))
            action = int(torch.argmax(logits, dim=-1).item())
            actions[i] = action
            account.execute(action, price)
    return actions


def _first_threshold_outcome(values: np.ndarray, threshold: float) -> str:
    for value in values:
        if value >= threshold:
            return "up_first"
        if value <= -threshold:
            return "down_first"
    return "neither"


def extreme_diagnostics(
    frame: pd.DataFrame,
    *,
    horizons: Iterable[int],
    threshold: float,
) -> dict:
    close = pd.to_numeric(frame["close"], errors="coerce").to_numpy(dtype=float)
    output: dict[str, dict] = {}
    for signal, expected_reversal in (
        ("ext_crash_signal", "up_first"),
        ("ext_surge_signal", "down_first"),
    ):
        flags = _as_bool(frame.get(signal, pd.Series(False, index=frame.index))).to_numpy()
        starts = np.flatnonzero(flags & ~np.r_[False, flags[:-1]])
        by_horizon: dict[str, dict] = {}
        for horizon in horizons:
            outcomes = {"up_first": 0, "down_first": 0, "neither": 0}
            final_returns: list[float] = []
            for index in starts:
                end = min(len(close), index + int(horizon) + 1)
                if end <= index + 1 or close[index] <= 0:
                    continue
                returns = close[index + 1 : end] / close[index] - 1.0
                outcome = _first_threshold_outcome(returns, threshold)
                outcomes[outcome] += 1
                final_returns.append(float(returns[-1]))
            total = sum(outcomes.values())
            by_horizon[str(int(horizon))] = {
                "episodes": total,
                "up_first": outcomes["up_first"],
                "down_first": outcomes["down_first"],
                "neither": outcomes["neither"],
                "expected_reversal": expected_reversal,
                "expected_reversal_rate_pct": outcomes[expected_reversal] / max(total, 1) * 100.0,
                "continuation_rate_pct": (
                    outcomes["down_first" if expected_reversal == "up_first" else "up_first"]
                    / max(total, 1)
                    * 100.0
                ),
                "median_final_return_pct": (
                    float(np.median(final_returns)) * 100.0 if final_returns else 0.0
                ),
            }
        output[signal] = {"episode_starts": int(len(starts)), "horizons": by_horizon}
    return output


def _regime_actions(frame: pd.DataFrame, model: np.ndarray, heuristic: np.ndarray) -> dict:
    regimes = frame.get("ext_regime_name", pd.Series("normal", index=frame.index)).fillna("normal")
    result: dict[str, dict] = {}
    for regime in sorted(regimes.astype(str).unique()):
        mask = regimes.astype(str).to_numpy() == regime
        result[regime] = {
            "bars": int(mask.sum()),
            "model_actions": {
                ACTION_NAMES[action]: int((model[mask] == action).sum()) for action in ACTION_NAMES
            },
            "heuristic_actions": {
                ACTION_NAMES[action]: int((heuristic[mask] == action).sum()) for action in ACTION_NAMES
            },
        }
    return result


def compare(
    *,
    market_csv: Path,
    predictions_csv: Path,
    initial_krw: float,
    bars_per_day: int,
    threshold: float,
    rl_checkpoint: Path | None = None,
) -> dict:
    market = pd.read_csv(market_csv)
    predictions = pd.read_csv(predictions_csv, usecols=["bar_index", "action"])
    predictions["bar_index"] = pd.to_numeric(predictions["bar_index"], errors="coerce")
    predictions["action"] = pd.to_numeric(predictions["action"], errors="coerce")
    predictions = predictions.dropna().astype({"bar_index": int, "action": int})
    frame = market.merge(predictions, on="bar_index", how="inner", validate="one_to_one")
    frame = frame.sort_values("bar_index").reset_index(drop=True)
    if frame.empty:
        raise ValueError("시장 CSV와 모델 예측 CSV가 겹치지 않습니다.")

    model_actions = frame["action"].to_numpy(dtype=np.int64)
    heuristic_actions = trend_heuristic_actions(frame)
    range_actions = minmax_actions(frame)
    hold_actions = buy_hold_actions(len(frame))
    prices = frame["close"].to_numpy(dtype=float)
    strategies = [
        _simulate(prices, model_actions, name="lora_model", initial_krw=initial_krw),
        _simulate(prices, heuristic_actions, name="trend_heuristic", initial_krw=initial_krw),
        _simulate(prices, range_actions, name="minmax_10_80", initial_krw=initial_krw),
        _simulate(prices, hold_actions, name="buy_and_hold", initial_krw=initial_krw),
    ]
    if rl_checkpoint is not None and rl_checkpoint.is_file():
        rl_actions = rl_policy_actions(frame, rl_checkpoint, initial_krw=initial_krw)
        strategies.insert(
            -1,
            _simulate(prices, rl_actions, name="rl_policy", initial_krw=initial_krw),
        )
    one_day = max(1, int(bars_per_day))
    horizons = sorted(set((60, one_day, one_day * 7)))
    return {
        "version": 1,
        "market_csv": str(market_csv.resolve()),
        "predictions_csv": str(predictions_csv.resolve()),
        "first_bar": int(frame.iloc[0]["bar_index"]),
        "last_bar": int(frame.iloc[-1]["bar_index"]),
        "processed_bars": len(frame),
        "bars_per_day_assumption": one_day,
        "assumed_market_days": len(frame) / one_day,
        "strategies": [result.__dict__ for result in strategies],
        "regime_actions": _regime_actions(frame, model_actions, heuristic_actions),
        "extreme_diagnostics": extreme_diagnostics(
            frame, horizons=horizons, threshold=float(threshold)
        ),
    }


def _markdown(payload: dict) -> str:
    lines = [
        "# LoRA 모델과 휴리스틱 비교",
        "",
        f"- 평가 봉: {payload['first_bar']}~{payload['last_bar']} ({payload['processed_bars']:,}개)",
        f"- 봉/일 가정: {payload['bars_per_day_assumption']:,}",
        f"- 환산 시장 기간: {payload['assumed_market_days']:.2f}일",
        "",
        "| 전략 | 청산 수익률 | 최대 낙폭 | 완료 거래 | 승률 |",
        "|---|---:|---:|---:|---:|",
    ]
    for row in payload["strategies"]:
        lines.append(
            f"| {row['name']} | {row['liquidation_return_pct']:+.3f}% | "
            f"{row['max_drawdown_pct']:.3f}% | {row['completed_trades']} | "
            f"{row['win_rate_pct']:.1f}% |"
        )
    lines += [
        "",
        "## 해석 주의",
        "",
        "- CSV에 시각 열이 없어 시장 기간은 봉/일 가정에 따른 환산값이다.",
        "- 폭락·폭등 사후 진단은 미래 가격을 평가에만 사용하며 휴리스틱 주문에는 사용하지 않는다.",
        "- 승률이 높아도 미청산 포지션 손실 때문에 전체 청산 수익률은 음수가 될 수 있다.",
    ]
    return "\n".join(lines) + "\n"


def main() -> None:
    parser = argparse.ArgumentParser(description="LoRA/휴리스틱/단순보유 동일 구간 비교")
    parser.add_argument("--predictions", type=Path, required=True)
    parser.add_argument("--csv", type=Path, default=Path("cache/market_extreme_labels.csv"))
    parser.add_argument("--output", type=Path, required=True)
    parser.add_argument("--initial-krw", type=float, default=DEFAULT_INITIAL_KRW)
    parser.add_argument(
        "--bars-per-day",
        type=int,
        default=288,
        help="CSV에 시각 열이 없으므로 명시하는 하루 봉 수(기본 5분봉 가정)",
    )
    parser.add_argument("--threshold", type=float, default=0.01, help="반전/연속 판정 가격 변화")
    parser.add_argument(
        "--rl-checkpoint",
        type=Path,
        default=Path("cache/rl_chart_policy_with_labels.pt"),
        help="존재하면 동일 구간에서 비교할 17특성 RL 정책",
    )
    args = parser.parse_args()
    payload = compare(
        market_csv=args.csv,
        predictions_csv=args.predictions,
        initial_krw=args.initial_krw,
        bars_per_day=args.bars_per_day,
        threshold=args.threshold,
        rl_checkpoint=args.rl_checkpoint,
    )
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
    report = args.output.with_suffix(".md")
    report.write_text(_markdown(payload), encoding="utf-8")
    print(_markdown(payload), end="")
    print(f"JSON: {args.output.resolve()}")
    print(f"보고서: {report.resolve()}")


if __name__ == "__main__":
    main()