예제 소스

surge_paper_train_loop.py

# -*- coding: utf-8 -*-
"""
장 개시~장 마감: `surge_paper_trade` 가상매매 → 장 마감 시 프로세스 종료
장 마감 후: 당일 `surge_paper_trades_YYYY-MM-DD.csv` 로 분석 JSONL 생성 → LoRA 학습
→ 다시 1) 로 무한 반복 (Ctrl-C 로 전체 종료)

사용법 (chart 디렉터리에서)

  python surge_paper_train_loop.py -- \\
    --interval-sec 60 --top-n 150

옵션 (루프 전용, `--` 앞에 둠)

  --train-data PATH     기본 training/data/surge_analysis.jsonl
  --lora-out PATH       기본 training/outputs/surge_lora
  --base-model NAME     기본 환경변수 BASE_MODEL 또는 Qwen/Qwen2.5-0.5B-Instruct
  --epochs N            train_surge_lora 에 전달 (기본 4)
  --skip-train          데이터셋만 만들고 학습 생략
  --skip-dataset        학습만 (CSV 없으면 스킵)

`--` 뒤 인자는 그대로 `surge_paper_trade.py` 에 넘깁니다 (단, 러너가 `--exit-after-session` 을 붙입니다).
"""

from __future__ import annotations

import argparse
import os
import subprocess
import sys
from datetime import date
from pathlib import Path

HERE = Path(__file__).resolve().parent


def _split_argv(argv: list[str]) -> tuple[list[str], list[str]]:
    if "--" in argv:
        i = argv.index("--")
        return argv[:i], argv[i + 1 :]
    return [], argv


def _run(cmd: list[str], *, label: str) -> int:
    print(f"\n=== {label} ===\n{' '.join(cmd)}\n", flush=True)
    r = subprocess.run(cmd, cwd=str(HERE))
    return int(r.returncode)


def main() -> int:
    raw = sys.argv[1:]
    loop_argv, paper_argv = _split_argv(raw)

    lp = argparse.ArgumentParser(add_help=True, description="장중 매매 ↔ 장마감 LoRA 반복")
    lp.add_argument(
        "--train-data",
        type=str,
        default=str(HERE / "training" / "data" / "surge_analysis.jsonl"),
        help="build_surge_dataset 출력 JSONL",
    )
    lp.add_argument(
        "--lora-out",
        type=str,
        default=str(HERE / "training" / "outputs" / "surge_lora"),
        help="LoRA 저장 디렉터리",
    )
    lp.add_argument(
        "--base-model",
        type=str,
        default=os.environ.get("BASE_MODEL", "Qwen/Qwen2.5-0.5B-Instruct"),
        help="train_surge_lora --base-model",
    )
    lp.add_argument("--epochs", type=float, default=4.0)
    lp.add_argument("--skip-train", action="store_true")
    lp.add_argument("--skip-dataset", action="store_true")
    args = lp.parse_args(loop_argv)

    train_data = Path(args.train_data)
    lora_out = Path(args.lora_out)
    py = sys.executable

    round_id = 0
    try:
        while True:
            round_id += 1
            print(f"\n{'#' * 60}\n# 라운드 {round_id}  {date.today()}\n{'#' * 60}\n", flush=True)

            paper_cmd = [py, str(HERE / "surge_paper_trade.py"), "--exit-after-session", *paper_argv]
            rc = _run(paper_cmd, label="가상매매 (장 마감 시까지)")
            if rc != 0:
                print(f"[warn] surge_paper_trade 종료 코드 {rc} — 잠시 후 재시도 (60s)", flush=True)
                import time

                time.sleep(60)
                continue

            day_csv = HERE / f"surge_paper_trades_{date.today():%Y-%m-%d}.csv"
            dataset_ok = True
            if not args.skip_dataset:
                if not day_csv.is_file():
                    print(f"[skip] 당일 거래 CSV 없음: {day_csv.name}", flush=True)
                    dataset_ok = False
                else:
                    ds_cmd = [
                        py,
                        str(HERE / "training" / "build_surge_dataset.py"),
                        "--csv",
                        str(day_csv),
                        "--out",
                        str(train_data),
                    ]
                    rc_ds = _run(ds_cmd, label="분석 데이터셋 생성")
                    if rc_ds != 0:
                        print(f"[warn] 데이터셋 생성 실패 코드 {rc_ds}", flush=True)
                        dataset_ok = False

            if not args.skip_train:
                if not train_data.is_file():
                    print(f"[skip] 학습용 JSONL 없음: {train_data}", flush=True)
                elif args.skip_dataset or dataset_ok:
                    train_cmd = [
                        py,
                        "-m",
                        "training",
                        "train_surge_lora.py",
                        "--data",
                        str(train_data),
                        "--out",
                        str(lora_out),
                        "--base-model",
                        args.base_model,
                        "--epochs",
                        str(args.epochs),
                    ]
                    rc_tr = _run(train_cmd, label="LoRA 학습")
                    if rc_tr != 0:
                        print(f"[warn] LoRA 학습 실패 코드 {rc_tr}", flush=True)

            print("\n[ok] 라운드 완료 — 다음 거래일 장 개시까지 surge_paper_trade 가 대기합니다.\n", flush=True)

    except KeyboardInterrupt:
        print("\n[stop] 전체 루프 종료 (Ctrl-C)", flush=True)
        return 0


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