openclaw/agents/stock/workspace/sim/__main__.py

"""sim CLI.

  python3 -m sim scan [--force]   # 1회 스캔 (launchd/cron 진입점)
  python3 -m sim status           # 가상계좌 요약
  python3 -m sim report [N]       # 최근 거래 N건 (기본 20)
  python3 -m sim reset [--yes]    # 가상계좌 초기화
  python3 -m sim variants sync    # 비교군을 백테스트 유형별 1위로 동기화 (수동 전용)
"""
from __future__ import annotations

import json
import sys

from . import config, engine
from .portfolio import Portfolio


def _cmd_scan(argv):
    snap = engine.scan_all(force='--force' in argv)
    if snap.get('skipped'):
        print(f"[skip] {snap['reason']} @ {snap['at']}")
        return
    s = snap['summary']
    print(f"자산 {s['equity']:,}원 ({s['total_return_pct']:+.2f}%) · 현금 {s['cash']:,} · "
          f"보유 {s['open_positions']} · 매수 {snap['counts']['buys']} 매도 {snap['counts']['sells']}")


def _cmd_status(argv):
    pf = Portfolio.load()
    s = pf.summary()
    print(f"가상계좌 (초기 {s['initial']:,}원)")
    print(f"  평가자산 {s['equity']:,}원  ({s['total_return_pct']:+.2f}%)")
    print(f"  현금 {s['cash']:,} · 주식 {s['held_value']:,}")
    print(f"  실현손익 {s['realized_pnl']:,} · 청산 {s['closed_trades']}건 "
          f"승률 {s['win_rate_pct']}% · MDD {s['max_drawdown_pct']}%")
    if pf.positions:
        print('  보유:')
        for p in pf.positions.values():
            cur = p.get('cur_price', p['entry_price'])
            pnl = (cur / p['entry_price'] - 1) * 100
            tr = ' [트레일링]' if p.get('trailing_on') else ''
            print(f"    {p['name']:16s} {p['qty']}주 @{p['entry_price']:,} → {cur:,} "
                  f"({pnl:+.1f}%) 손절 {p['stop']:,} 목표 {p['target']:,}{tr}")


def _cmd_report(argv):
    n = next((int(a) for a in argv if a.isdigit()), 20)
    if not config.TRADES_PATH.exists():
        print('거래 없음')
        return
    lines = config.TRADES_PATH.read_text().splitlines()[-n:]
    for line in lines:
        r = json.loads(line)
        if r['side'] == 'BUY':
            print(f"{r['ts'][:16]} 매수 {r['name']:16s} {r['qty']}주 @{r['price']:,} · {r['reason']}")
        else:
            print(f"{r['ts'][:16]} 매도 {r['name']:16s} {r['qty']}주 @{r['price']:,} "
                  f"손익 {r['pnl']:,}({r['pnl_pct']:+.1f}%) · {r['reason']}")


def _cmd_sweep(argv):
    from . import sweep
    frac = next((float(a) for a in argv if a.replace('.', '').isdigit()), 0.7)
    res = sweep.run_sweep(train_frac=frac)
    if res.get('error'):
        print('스윕 실패:', res)
        return
    print(f"스윕 완료 — {res['count']}개 조합 · 검증 {res['split']['test']} "
          f"· 수급반영 {res['flow_used']}")
    for r in res['results'][:10]:
        p, te = r['params'], r['test']
        print(f"  RR{p['RR_RATIO']} S{p['STOP_ATR_MULT']} SMA{p['SMA_LONG']} P{p['PULLBACK_ATR_MULT']} → "
              f"검증 {te['total_return_pct']:+}% MDD {te['mdd_pct']}% 거래 {te['trades']} 승률 {te['win_rate_pct']}%")


def _cmd_backfill(argv):
    from . import backfill_flow
    pages = int(argv[argv.index('--pages') + 1]) if '--pages' in argv else backfill_flow.DEFAULT_PAGES
    res = backfill_flow.backfill(pages=pages, force='--force' in argv)
    print(f"수급 백필 — 신규 {res['fetched']} · 스킵 {res['skipped']} · 실패 {res['failed']} "
          f"· DB {res['codes_in_db']}종목 {res['total_rows']}행")


def _cmd_backtest(argv):
    from . import backtest, universe
    codes = [e['code'] for e in universe.build_universe()]
    import json
    print(json.dumps(backtest.run(None, codes), ensure_ascii=False, indent=2))


def _cmd_backfill_index(argv):
    from . import benchmark
    pages = int(argv[argv.index('--pages') + 1]) if '--pages' in argv else 8
    res = benchmark.backfill(pages=pages)
    print(f"지수 백필 — 신규 {res}")
    for idx in benchmark.INDICES:
        ks = sorted(benchmark.series(idx))
        if ks:
            print(f"  {idx}: {ks[0]} ~ {ks[-1]} ({len(ks)}일)")


def _cmd_variants(argv):
    from . import variants
    sub = argv[0] if argv else 'list'
    if sub == 'seed':
        n = next((int(a) for a in argv[1:] if a.isdigit()), 3)
        vs = variants.seed_from_sweep(n)
        print(f'변이 {len(vs)}개 등록:')
        for v in vs:
            print(' ', v['id'], v['name'])
    elif sub == 'sync':
        import json as _json
        from . import config as _cfg
        res = _json.loads((_cfg.STATE_DIR / 'backtest_results.json').read_text())
        print('동기화:', variants.sync_label_tops(res))
    elif sub == 'reset':
        variants.reset_all()
        print('변이 계좌 초기화 (정의 유지)')
    elif sub == 'clear':
        variants.clear()
        print('변이 정의·계좌 전부 삭제')
    else:
        vs = variants.load_variants()
        print(f'변이 {len(vs)}개:')
        for v in vs:
            print(' ', v['id'], v['name'], v['params'])


def _cmd_reset(argv):
    if '--yes' not in argv:
        print('정말 초기화하려면 --yes 를 붙이세요 (가상계좌·거래·스냅샷 삭제)')
        return
    for p in (config.PORTFOLIO_PATH, config.TRADES_PATH, config.LAST_SCAN_PATH):
        if p.exists():
            p.unlink()
    print(f'초기화 완료 — 가상자본 {config.INITIAL_CAPITAL:,}원')


def main():
    argv = sys.argv[1:]
    cmd = argv[0] if argv else 'status'
    rest = argv[1:]
    {'scan': _cmd_scan, 'status': _cmd_status, 'report': _cmd_report,
     'sweep': _cmd_sweep, 'backtest': _cmd_backtest, 'backfill': _cmd_backfill,
     'backfill-index': _cmd_backfill_index,
     'variants': _cmd_variants, 'reset': _cmd_reset}.get(cmd, lambda a: print(__doc__))(rest)


if __name__ == '__main__':
    main()