auto-stock/backend/_sweep_single_axis.py

#!/usr/bin/env python3
"""단일 파라미터 축 민감도 스윕 (run_backtest.py의 'normal' 모드 필드 1개만 오버라이드).

run_backtest_sweep.py와 동일한 텍스트 치환 방식을 사용하되, ADAPTIVE_MODES 전체가
아니라 normal 모드의 지정된 키 하나만 바꿔서 단일 축 민감도를 측정한다.
"""
import subprocess, json, sys, os, time, argparse

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
BACKTEST_SCRIPT = os.path.join(BASE_DIR, "run_backtest.py")

BASELINE_MODES = {
    "aggressive": {"rsi_entry": 48.0, "bear_rsi_entry": 38.0, "bb_pct_b_scale": 0.45, "position_size": 0.07,
                   "max_holdings": 15, "stop_loss": -8.0, "stop_loss_vol": -12.0, "atr_stop_mult": 3.0,
                   "atr_trail_mult": 1.8, "rsi_partial": 58.0, "rsi_full": 72.0},
    "normal":     {"rsi_entry": 43.0, "bear_rsi_entry": 32.0, "bb_pct_b_scale": 0.35, "position_size": 0.05,
                   "max_holdings": 12, "stop_loss": -6.0, "stop_loss_vol": -10.0, "atr_stop_mult": 2.5,
                   "atr_trail_mult": 1.5, "rsi_partial": 52.0, "rsi_full": 67.0},
    "defensive":  {"rsi_entry": 35.0, "bear_rsi_entry": 25.0, "bb_pct_b_scale": 0.20, "position_size": 0.03,
                   "max_holdings": 8, "stop_loss": -5.0, "stop_loss_vol": -8.0, "atr_stop_mult": 2.0,
                   "atr_trail_mult": 1.2, "rsi_partial": 48.0, "rsi_full": 60.0},
    "crisis":     {"rsi_entry": 28.0, "bear_rsi_entry": 20.0, "bb_pct_b_scale": 0.10, "position_size": 0.0,
                   "max_holdings": 3, "stop_loss": -4.0, "stop_loss_vol": -6.0, "atr_stop_mult": 1.5,
                   "atr_trail_mult": 1.0, "rsi_partial": 45.0, "rsi_full": 55.0},
}


def patch_and_run(modes, start, end, timeout=600):
    with open(BACKTEST_SCRIPT, "r") as f:
        content = f.read()
    start_marker = "ADAPTIVE_MODES = {"
    end_marker = "\n\ndef determine_adaptive_mode("
    s = content.find(start_marker)
    e = content.find(end_marker)
    if s == -1 or e == -1:
        print("[ERROR] ADAPTIVE_MODES 블록을 찾을 수 없습니다")
        return None
    new_block = "ADAPTIVE_MODES = " + json.dumps(modes, indent=4, ensure_ascii=False)
    patched = content[:s] + new_block + content[e:]

    tmp_script = os.path.join(BASE_DIR, f"_tmp_sweep_{os.getpid()}.py")
    with open(tmp_script, "w") as f:
        f.write(patched)
    try:
        result = subprocess.run(
            [sys.executable, tmp_script, "--start", start, "--end", end],
            capture_output=True, text=True, timeout=timeout,
        )
        output = result.stdout + result.stderr
    finally:
        try:
            os.remove(tmp_script)
        except OSError:
            pass

    metrics = {}
    for line in output.split("\n"):
        line = line.strip()
        if "총 수익률:" in line:
            metrics["return"] = line.split(":")[-1].strip()
        elif "CAGR:" in line:
            metrics["cagr"] = line.split(":")[-1].strip()
        elif "Max Drawdown:" in line:
            metrics["mdd"] = line.split(":")[-1].strip()
        elif "승률:" in line:
            metrics["winrate"] = line.split(":")[-1].strip()
        elif "총 매매 횟수:" in line:
            metrics["trades"] = line.split(":")[-1].strip()
        elif "Sharpe Ratio:" in line:
            metrics["sharpe"] = line.split(":")[-1].strip()
        elif "Profit Factor:" in line:
            metrics["pf"] = line.split(":")[-1].strip()
        elif "최대 연패:" in line:
            metrics["max_loss_streak"] = line.split(":")[-1].strip()
    if not metrics:
        print("---- RAW OUTPUT (no metrics parsed) ----")
        print(output[-3000:])
    return metrics


AXIS_KEY_MAP = {
    "rsi_entry": "rsi_entry",
    "stop_loss": "stop_loss",
    "rsi_partial": "rsi_partial",
    "atr_stop_mult": "atr_stop_mult",
}


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--axis", required=True, choices=list(AXIS_KEY_MAP.keys()))
    ap.add_argument("--values", required=True, help="comma separated values, e.g. 42,46,50")
    ap.add_argument("--start", default="2024-09-01")
    ap.add_argument("--end", default="2026-09-05")
    args = ap.parse_args()

    key = AXIS_KEY_MAP[args.axis]
    values = [float(v) for v in args.values.split(",")]

    print(f"=== Axis: {args.axis} ({key}) | period {args.start}~{args.end} ===")
    for v in values:
        import copy
        modes = copy.deepcopy(BASELINE_MODES)
        modes["normal"][key] = v
        print(f"\n--- {args.axis}={v} (normal mode only) ---")
        t0 = time.time()
        m = patch_and_run(modes, args.start, args.end)
        elapsed = time.time() - t0
        print(f"  elapsed={elapsed:.1f}s  {m}")


if __name__ == "__main__":
    main()