hyowons
145 lines
5.4 KiB
Python
145 lines
5.4 KiB
Python
"""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] # 가상계좌 초기화 (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 == '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()
|