hyowons
149 lines
5.1 KiB
Python
149 lines
5.1 KiB
Python
"""병렬 페이퍼 변이(variant) 관리 — 각 변이는 {이름, 파라미터} + 자체 가상계좌.
|
|
|
|
정의: state/sim/variants.json = [{"id":"v1","name":"...","params":{KEY:VAL,...}}]
|
|
계좌: state/sim/variants/<id>/{portfolio.json, trades.jsonl}
|
|
비교: state/sim/variants_compare.json (웹 비교 탭)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from . import config
|
|
|
|
VARIANTS_PATH = config.STATE_DIR / 'variants.json'
|
|
VARIANTS_DIR = config.STATE_DIR / 'variants'
|
|
COMPARE_PATH = config.STATE_DIR / 'variants_compare.json'
|
|
|
|
|
|
def load_variants() -> list[dict]:
|
|
try:
|
|
return json.loads(VARIANTS_PATH.read_text())
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
def save_variants(variants: list[dict]):
|
|
config.STATE_DIR.mkdir(parents=True, exist_ok=True)
|
|
VARIANTS_PATH.write_text(json.dumps(variants, ensure_ascii=False, indent=2))
|
|
|
|
|
|
def variant_paths(vid: str):
|
|
d = VARIANTS_DIR / vid
|
|
return d / 'portfolio.json', d / 'trades.jsonl'
|
|
|
|
|
|
def _label(params: dict) -> str:
|
|
short = {'RR_RATIO': 'RR', 'STOP_ATR_MULT': 'S', 'SMA_LONG': 'SMA',
|
|
'PULLBACK_ATR_MULT': 'P', 'VOLUME_BREAKOUT_MULT': 'V',
|
|
'TURNOVER_OVERHEAT_MULT': 'OH', 'RSI_OVERBOUGHT': 'RSI↑',
|
|
'RSI_OVERSOLD': 'RSI↓', 'MAX_HOLD_DAYS': 'HOLD', 'RISK_PER_TRADE_PCT': 'RISK'}
|
|
return '·'.join(f'{short.get(k, k)}{v:g}' for k, v in params.items())
|
|
|
|
|
|
def seed_from_sweep(n: int = 3) -> list[dict]:
|
|
"""backtest_results.json 검증 상위 n개 → 변이로 등록 (기존 계좌는 유지)."""
|
|
try:
|
|
res = json.loads((config.STATE_DIR / 'backtest_results.json').read_text())
|
|
except Exception:
|
|
return []
|
|
existing = {v['id']: v for v in load_variants()}
|
|
variants = list(existing.values())
|
|
seen_params = {json.dumps(v['params'], sort_keys=True) for v in variants}
|
|
rank = 0
|
|
for r in res.get('results', []):
|
|
if rank >= n:
|
|
break
|
|
params = r['params']
|
|
key = json.dumps(params, sort_keys=True)
|
|
if key in seen_params:
|
|
continue
|
|
rank += 1
|
|
vid = f'v{len(variants)+1}'
|
|
variants.append({'id': vid, 'name': f'#{rank} {_label(params)}', 'params': params})
|
|
seen_params.add(key)
|
|
save_variants(variants)
|
|
return variants
|
|
|
|
|
|
def sync_label_tops(results: dict) -> dict:
|
|
"""sweep 결과에서 세부 유형(시장적합 라벨)별 1위를 비교군과 동기화 — 수동 추가/제거 불필요.
|
|
|
|
파라미터가 같은 변이는 계좌 유지, 빠진 유형 대표는 추가, 더는 대표가 아닌 변이는 삭제.
|
|
(2026-06-10 기준전략 폐지 — 메인 제외 규칙 삭제, 유형별 1위 전부 변이로.)"""
|
|
|
|
def _k(p):
|
|
return json.dumps({k: float(v) for k, v in (p or {}).items()}, sort_keys=True)
|
|
|
|
def _sk(r):
|
|
w = r.get('worst_ret')
|
|
return (w if w is not None else -999,
|
|
r['test'].get('total_return_pct', -999) or -999)
|
|
|
|
from .sim_web import _market_fit_summary # 함수 레벨 import (순환 회피)
|
|
uniq = {}
|
|
for r in results.get('results', []):
|
|
uniq.setdefault(_k(r['params']), r)
|
|
by_label = {}
|
|
for r in uniq.values():
|
|
lbl = _market_fit_summary(r['params'])
|
|
if lbl not in by_label or _sk(r) > _sk(by_label[lbl]):
|
|
by_label[lbl] = r
|
|
desired = {_k(r['params']): r['params'] for r in by_label.values()}
|
|
current = {_k(v['params']): v for v in load_variants()}
|
|
removed = added = 0
|
|
for k, v in current.items():
|
|
if k not in desired:
|
|
remove_variant(v['id'])
|
|
removed += 1
|
|
for k, p in desired.items():
|
|
if k not in current:
|
|
if add_variant(p):
|
|
added += 1
|
|
return {'labels': len(by_label), 'added': added, 'removed': removed,
|
|
'kept': len(current) - removed}
|
|
|
|
|
|
def add_variant(params: dict) -> dict | None:
|
|
"""단일 params 조합을 변이로 추가 (백테스트 행 → 비교군). 이미 있으면 None."""
|
|
variants = load_variants()
|
|
seen = {json.dumps(v['params'], sort_keys=True) for v in variants}
|
|
if json.dumps(params, sort_keys=True) in seen:
|
|
return None
|
|
nums = [int(v['id'][1:]) for v in variants if v['id'][1:].isdigit()]
|
|
vid = f'v{(max(nums) + 1) if nums else 1}'
|
|
v = {'id': vid, 'name': _label(params), 'params': params}
|
|
variants.append(v)
|
|
save_variants(variants)
|
|
return v
|
|
|
|
|
|
def remove_variant(vid: str):
|
|
"""변이 1개 삭제 (정의 + 계좌)."""
|
|
import shutil
|
|
save_variants([v for v in load_variants() if v['id'] != vid])
|
|
d = VARIANTS_DIR / vid
|
|
if d.exists():
|
|
shutil.rmtree(d)
|
|
|
|
|
|
def reset_variant(vid: str):
|
|
"""변이 1개 성적만 초기화 (정의 유지 — 계좌 디렉터리 삭제 후 다음 스캔에 재생성)."""
|
|
import shutil
|
|
d = VARIANTS_DIR / vid
|
|
if d.exists():
|
|
shutil.rmtree(d)
|
|
|
|
|
|
def reset_all():
|
|
"""모든 변이 계좌 초기화 (variants.json 정의는 유지)."""
|
|
import shutil
|
|
if VARIANTS_DIR.exists():
|
|
shutil.rmtree(VARIANTS_DIR)
|
|
|
|
|
|
def clear():
|
|
"""변이 정의·계좌 전부 삭제."""
|
|
reset_all()
|
|
if VARIANTS_PATH.exists():
|
|
VARIANTS_PATH.unlink()
|