예제 소스
Qwen2.5-VL 차트 LoRA: PNG로 매수·매도·유지 학습하기 · 파일 59개
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()