kis_bot/scripts/mom_bt_one_day_trial120.py

#!/usr/bin/env python3
"""Apply trial#120 params → 1일 모멘텀 백테 (호가 ON). DB 미변경."""
from __future__ import annotations

import json
import sys
import time
from pathlib import Path

_ROOT = Path(__file__).resolve().parents[1]
if str(_ROOT) not in sys.path:
    sys.path.insert(0, str(_ROOT))


def main() -> int:
    t0 = time.time()
    start = end = "2026-08-25"
    out_json = _ROOT / "logs" / "mom_bt_20260825_trial120_ob_on.json"
    print(f"start prepare {start} OB=on", flush=True)

    from database import TradeDB
    from kis_trader.backtest.optuna_momentum import prepare_momentum_search_context
    from kis_trader.backtest.param_search_momentum import evaluate_momentum_param_combo

    db = TradeDB()
    row = db.conn.execute(
        "SELECT payload_json FROM optuna_study_result WHERE study_name=%s",
        ("momentum_tpe_20260818_20260821_20260823_192516",),
    ).fetchone()
    payload = json.loads(row["payload_json"])
    t120 = next(x for x in payload["results_gated"] if x.get("optuna_trial_number") == 120)
    combo = dict(t120["params"])
    print(
        "trial120 ob=%s whip=%s sl=%s tp=%s" % (
            combo.get("_orderbook_filter_enabled"),
            combo.get("whipsaw_filter_enabled"),
            combo.get("sl_pct"),
            combo.get("tp_pct"),
        ),
        flush=True,
    )

    ctx = prepare_momentum_search_context(
        start,
        end,
        "tpe",
        orderbook_filter="on",
        market="KR",
    )
    if ctx is None:
        print("PREPARE FAILED", flush=True)
        return 1

    print(
        "ctx ready codes=%d ticks=%s ob_codes=%d fee=%s slot=%s max=%s budget=%s"
        % (
            len(ctx.codes_candles),
            bool(ctx.ticks_by_code),
            len(ctx.orderbook_by_code or {}),
            ctx.fee_rate,
            ctx.slot_money,
            ctx.max_stocks,
            ctx.total_budget_krw,
        ),
        flush=True,
    )

    base = dict(ctx.base_fixed)
    base["_orderbook_filter_enabled"] = True
    base["ob_filter_enabled"] = True

    grid_keys = list(payload.get("grid_keys") or [])
    if not grid_keys:
        grid_keys = [k for k in combo.keys() if not str(k).startswith("_")]

    result = evaluate_momentum_param_combo(
        combo,
        base_fixed=base,
        grid_keys=grid_keys,
        codes_candles=ctx.codes_candles,
        min_trades=0,
        min_win_rate=0.0,
        min_pf=0.0,
        universe_by_slot=ctx.universe_by_slot,
        slot_money=ctx.slot_money,
        max_stocks=ctx.max_stocks,
        total_budget_krw=ctx.total_budget_krw,
        fee_rate=ctx.fee_rate,
        sell_tax=ctx.sell_tax,
        period_days=1,
        cache_holder=ctx.cache_holder,
        ticks_by_code=ctx.ticks_by_code,
        orderbook_by_code=ctx.orderbook_by_code,
        program_by_code=getattr(ctx, "program_by_code", None),
        start_key="20260825",
        end_key="20260825",
        include_trades=True,
    )
    print("elapsed_sec", round(time.time() - t0, 1), flush=True)
    if not result:
        print("NO RESULT", flush=True)
        return 2

    trades = result.get("_trades") or result.get("trades") or []
    out = {
        "ok": True,
        "date": start,
        "source": "optuna trial#120 + prepare orderbook_filter=on",
        "study": "momentum_tpe_20260818_20260821_20260823_192516",
        "total_trades": result.get("total_trades"),
        "win_rate": result.get("win_rate"),
        "total_pnl": result.get("total_pnl"),
        "pf": result.get("pf"),
        "trades": trades,
    }
    print(
        "RESULT trades=%s WR=%s PnL=%s PF=%s"
        % (out["total_trades"], out["win_rate"], out["total_pnl"], out["pf"]),
        flush=True,
    )
    for tr in trades:
        code = str(tr.get("code") or "")
        line = (
            "  %s->%s %s %s pnl=%s entry=%s exit=%s"
            % (
                tr.get("buy_time") or tr.get("entry_time"),
                tr.get("sell_time") or tr.get("exit_time"),
                code,
                tr.get("sell_reason"),
                tr.get("pnl"),
                tr.get("entry_price") or tr.get("buy_price"),
                tr.get("exit_price") or tr.get("sell_price"),
            )
        )
        print(line, flush=True)
        if code == "417010":
            print("  *** NANOTEAM IN BACKTEST ***", flush=True)

    out_json.write_text(json.dumps(out, ensure_ascii=False, indent=2, default=str))
    print("JSON", out_json, flush=True)
    print("DONE", flush=True)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())