"""시뮬 대시보드 — 별도 읽기전용 웹서버 (포트 18792).
sim state 파일(last_scan.json·portfolio.json·trades.jsonl)만 읽어 렌더. 키움 호출 없음.
외부 노출은 NAS reverse proxy(sim.hyowons.net → mac:18792) 경유. Tailnet 내부망 한정, 인증 없음.
python3 -m sim.sim_web serve # launchd 상시
python3 -m sim.sim_web render # 1회 HTML stdout (디버깅)
"""
from __future__ import annotations
import html
import json
import subprocess
import sys
import threading
import time
import urllib.parse
import urllib.request
from datetime import datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from . import config
from .portfolio import Portfolio
BIND_HOST = ''
BIND_PORT = 18792
STATE_BADGE = {
'SELL': ('매도', '#ef4444'), 'ADD': ('추격매수', '#16a34a'), 'BUY': ('매수', '#22c55e'),
'HOLD': ('보유', '#3b82f6'), 'WAIT': ('대기', '#eab308'), 'SKIP': ('제외', '#6b7280'),
'NODATA': ('데이터없음', '#374151'),
}
STATE_ORDER = ['HOLD', 'ADD', 'BUY', 'SELL', 'WAIT', 'SKIP', 'NODATA']
ORIGIN_LABEL = {'watch': '워치', 'interest': '관심', 'held': '보유', 'manual': '수동'}
# 모바일 당겨서 새로고침 (behive_web 방식 차용 — 별도 상수로 템플릿 f-string 중괄호 충돌 회피)
# #page를 손가락 따라 translateY, #ptr 원형(화살표→스피너), resistance 0.55·임계 70px. 커밋 시 reload.
PTR_HTML = """
"""
# 탭 복원(?t=) + 저장 배너(?saved=1) + 자동새로고침(튜닝 탭/입력 중엔 멈춤, 현재 탭 유지)
REFRESH_JS = """
"""
# 관심종목 탭 종목 검색 — 입력(디바운스)·검색 버튼·Enter로 /api/stock_search 후보 칩. 칩 클릭=폼 POST로 추가.
_WATCH_SEARCH_JS = """
"""
# 백테스트 순위표 클라이언트 정렬 — 행의 data-* 수치로 재배치. 상태는 sessionStorage 유지.
# 기본 방향: best-first (검증·학습·승률·손익비 내림차순 / MDD 오름차순). 같은 항목 재클릭 시 토글.
_BT_SORT_JS = """
"""
def _load(path, default):
try:
return json.loads(path.read_text())
except Exception:
return default
# ---- 스윕(백테스트 재계산) 백그라운드 트리거 ----
# sweep은 ~80초라 동기 응답 불가 → subprocess 백그라운드 + 진행상태 파일로 표시.
SWEEP_STATUS = config.STATE_DIR / 'sweep_status.json'
_WORKSPACE = config.STATE_DIR.parent.parent # `-m sim` 실행 cwd
def _sweep_status() -> dict:
return _load(SWEEP_STATUS, {})
def _write_sweep_status(d: dict):
config.STATE_DIR.mkdir(parents=True, exist_ok=True)
SWEEP_STATUS.write_text(json.dumps(d, ensure_ascii=False))
def _trigger_sweep():
"""실행 중이 아니면 백그라운드로 `python3 -m sim sweep`. 진행상태를 파일에 남김."""
st = _sweep_status()
if st.get('state') == 'running':
try: # 5분 넘게 running이면 stale로 보고 재시작 허용
started = datetime.fromisoformat(st.get('started', ''))
if (datetime.now(config.KST) - started).total_seconds() < 300:
return
except Exception:
pass
_write_sweep_status({'state': 'running', 'started': datetime.now(config.KST).isoformat()})
def _run():
try:
p = subprocess.Popen([sys.executable, '-m', 'sim', 'sweep'], cwd=str(_WORKSPACE),
stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
_, err = p.communicate()
if p.returncode == 0:
_write_sweep_status({'state': 'done', 'finished': datetime.now(config.KST).isoformat()})
else:
_write_sweep_status({'state': 'error', 'finished': datetime.now(config.KST).isoformat(),
'msg': (err or b'').decode('utf-8', 'ignore')[-300:]})
except Exception as e:
_write_sweep_status({'state': 'error', 'finished': datetime.now(config.KST).isoformat(),
'msg': str(e)})
threading.Thread(target=_run, daemon=True).start()
def _check_mark(ok):
return {True: '✓', False: '✗', None: '·'}.get(ok, '·')
def _check_color(ok):
return {True: '#22c55e', False: '#ef4444', None: '#9ca3af'}.get(ok, '#9ca3af')
def _esc(v):
return html.escape(str(v if v is not None else ''))
def _num(s):
"""폼 문자열 → 정수면 int, 아니면 float (변이 params 타입 보존)."""
f = float(s)
return int(f) if f == int(f) else f
def _pkey(params) -> str:
"""params 비교 키 — 값을 float로 정규화 (int 3 vs float 3.0 매칭 일치)."""
return json.dumps({k: float(v) for k, v in (params or {}).items()}, sort_keys=True)
_PSHORT = {'SMA_LONG': 'SMA', 'RR_RATIO': 'RR', 'STOP_ATR_MULT': 'S', 'PULLBACK_ATR_MULT': 'P',
'VOLUME_BREAKOUT_MULT': 'V', 'TURNOVER_BREAKOUT_MULT': 'TB', 'TURNOVER_OVERHEAT_MULT': 'OH'}
def _plabel(params) -> str:
return '·'.join(f'{_PSHORT.get(k, k)}{v:g}' for k, v in (params or {}).items())
def _chips(checks) -> str:
"""신호 체크리스트 → 칩 HTML (판단근거 표시 공용)."""
return ''.join(
f''
f'{_check_mark(c.get("ok"))} {_esc(c.get("label"))}{_esc(c.get("detail"))}'
for c in (checks or [])
)
def _kpi(label, value, sub=''):
sub_h = f'
{_esc(sub)}
' if sub else ''
return f'
{_esc(label)}
{_esc(value)}
{sub_h}
'
def _signed(v, suffix='억'):
c = '#22c55e' if (v or 0) > 0 else '#ef4444' if (v or 0) < 0 else '#9ca3af'
return f'{(v or 0):+,}{suffix}'
_idx_cache = {'data': {}, 'ts': 0.0}
_IDX_TTL = 60.0
def _fetch_naver_index(code: str) -> dict | None:
"""네이버 지수 1개 (코스피/코스닥) closePrice·등락. 실패 시 None."""
try:
url = f'https://m.stock.naver.com/api/index/{code}/basic'
req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
with urllib.request.urlopen(req, timeout=3.0) as resp:
data = json.loads(resp.read().decode('utf-8', 'ignore'))
price = str(data.get('closePrice') or '').replace(',', '')
if not price:
return None
return {'price': float(price),
'change': float(str(data.get('compareToPreviousClosePrice') or '0').replace(',', '')),
'pct': float(str(data.get('fluctuationsRatio') or '0').replace(',', ''))}
except Exception as e:
sys.stderr.write(f'[sim_web] index {code} fetch 실패: {e}\n')
return None
def _get_indices() -> dict:
"""KOSPI·KOSDAQ 지수값 (in-process 60s 캐시 — sim_web은 상시 데몬)."""
now = time.time()
if _idx_cache['data'] and now - _idx_cache['ts'] < _IDX_TTL:
return _idx_cache['data']
out = {}
for code in ('KOSPI', 'KOSDAQ'):
r = _fetch_naver_index(code)
if r:
out[code] = r
if out:
_idx_cache['data'] = out
_idx_cache['ts'] = now
return out
def _market_panel() -> str:
"""시장 탭 — market_indicators_history.jsonl에서 시장별 ADR·상승하락·투자자 순매수."""
path = config.MARKET_HISTORY
if not path.exists():
return '
시장 데이터 없음
'
by_market: dict[str, list] = {}
try:
for line in path.read_text().splitlines():
if line.strip():
r = json.loads(line)
by_market.setdefault(r.get('market', ''), []).append(r)
except Exception:
return '
시장 데이터 읽기 실패
'
idx = _get_indices()
order = ['KOSPI', 'KOSDAQ']
keys = [k for k in order if k in by_market] + [k for k in by_market if k not in order]
out = ''
for k in keys:
recs = by_market[k]
r = recs[-1]
ix = idx.get(k)
if ix:
ic = '#22c55e' if ix['change'] > 0 else '#ef4444' if ix['change'] < 0 else '#9ca3af'
idx_row = (f'
')
else:
idx_row = ''
label = _esc(r.get('market_label') or k)
rise, fall, steady = r.get('rise', 0), r.get('fall', 0), r.get('steady', 0)
total = rise + fall + steady
breadth = (rise / total * 100) if total else 0
ok = total and (rise / total) >= config.MARKET_BREADTH_MIN
vlabel, vcolor = ('양호', '#22c55e') if ok else ('약세', '#ef4444')
adr = r.get('adr')
adr_s = f'{adr:.1f}' if isinstance(adr, (int, float)) else '—'
recent = [rr.get('adr') for rr in recs[-5:] if isinstance(rr.get('adr'), (int, float))]
recent_s = ' → '.join(f'{a:.0f}' for a in recent) if recent else '—'
out += (
f'
')
return out + note
def _tune_panel() -> str:
"""파라미터 튜닝 폼 — 그룹별 number 입력. 저장 시 params.json 반영(다음 스캔부터)."""
eff = config.load_params()
groups: dict[str, list] = {}
for spec in config.TUNABLE:
groups.setdefault(spec['group'], []).append(spec)
body = ''
''
'
저장하면 다음 스캔(최대 15분 뒤)부터 적용 · '
'범위 벗어난 값은 자동 보정
')
return body
def _alpha_inline(bench: dict) -> str:
"""benchmark dict({KOSPI/KOSDAQ:{pct,alpha}}) → '코스피 +5.2%p · 코스닥 -1.1%p' 문자열."""
parts = []
for idx, lbl in (('KOSPI', '코스피'), ('KOSDAQ', '코스닥')):
a = (bench.get(idx) or {}).get('alpha')
if a is None:
continue
c = '#22c55e' if a >= 0 else '#ef4444'
parts.append(f'{lbl} {a:+g}%p')
return ' · '.join(parts) if parts else '지수 데이터 없음'
def _bt_alpha_row(res: dict) -> str:
bench = res.get('benchmark') or {}
return (f'
지수대비(알파)'
f'{_alpha_inline(bench)}
')
def _backtest_panel() -> str:
"""백테스트 스윕 순위표 — 검증기간(out-of-sample) 성적순. 행마다 '이 튜닝 적용'."""
res = _load(config.STATE_DIR / 'backtest_results.json', None)
if not res or not res.get('results'):
return ('
'
)
from . import variants as vmod
registered = {_pkey(v['params']): v['id'] for v in vmod.load_variants()}
main_p = config.load_params()
main_pkey = _pkey({k: main_p[k] for k in res.get('swept_params', []) if k in main_p})
rows = ''
for i, r in enumerate(res['results'][:40]):
p, te, tr = r['params'], r['test'], r['train']
hidden = ''.join(f'' for k, v in p.items())
pk = _pkey(p)
vid = registered.get(pk)
is_main = pk == main_pkey
main_badge = ' 메인' if is_main else ''
cmp_badge = (' 비교군'
if vid else '')
if vid: # 비교군 등록됨 → 제거 (메인 중복이라도 정리 가능)
cmp_btn = (f'')
elif is_main: # 메인은 비교 탭에 항상 포함 → 추가 불가
cmp_btn = '
메인 — 비교 탭에 항상 포함
'
else:
cmp_btn = (f'')
teret = te.get('total_return_pct', 0)
trret = tr.get('total_return_pct', 0)
tc = '#22c55e' if teret >= 0 else '#ef4444'
overfit = (trret - teret) > 40 # 학습≫검증 → 과최적 의심
of_tag = ' 과최적?' if overfit else ''
rank = f'#{i+1}'
# 정렬용 수치 (없는 값은 최하위로 — MDD는 낮을수록 좋아 큰값, 나머지는 작은값)
mdd, wr, pf = te.get('mdd_pct'), te.get('win_rate_pct'), te.get('profit_factor')
d_te = teret if isinstance(teret, (int, float)) else -999
d_tr = trret if isinstance(trret, (int, float)) else -999
d_mdd = mdd if isinstance(mdd, (int, float)) else 9999
d_wr = wr if isinstance(wr, (int, float)) else -1
d_pf = 9e9 if pf == 'inf' else (pf if isinstance(pf, (int, float)) else -1)
rows += (
f'{rank} '
f'{_esc(_plabel(p))}{main_badge}{cmp_badge}'
f'검증 {teret:+g}%'
f'
'
f'
검증 수익 / 학습 수익'
f'{teret:+g}% / {trret:+g}%{of_tag}
'
f'{_bt_alpha_row(te)}'
f'
검증 MDD{mdd}%
'
f'
검증 거래수 / 승률{te.get("trades")}건 · {wr}%
'
f'
검증 손익비(PF){pf}
'
f'
'
f''
f'{cmp_btn}'
f'
'
)
return head + ctrl + sortbar + f'
{rows}
' + _BT_SORT_JS
def _compare_panel() -> str:
"""병렬 페이퍼 비교 — 메인 + 변이들의 실시간 누적 성적.
변이 정의(variants.json) + 각 계좌(portfolio.json)에서 직접 읽어 렌더 →
추가/초기화 버튼이 다음 스캔을 기다리지 않고 즉시 반영된다.
"""
from . import variants as vmod, benchmark
defs = vmod.load_variants()
today = datetime.now(config.KST).strftime('%Y%m%d')
def _row(rid, name, params, pf):
s = pf.summary()
return {'id': rid, 'name': name, 'params': params, 'summary': s,
'benchmark': benchmark.compare(s.get('total_return_pct'),
benchmark.norm_date(pf.created_at), today)}
# 메인 + 변이 계좌를 소스에서 직접 수집
rows_data = [_row('main', '메인 (현재 튜닝)', config.load_params(), Portfolio.load())]
for v in defs:
pp, tp = vmod.variant_paths(v['id'])
rows_data.append(_row(v['id'], v.get('name', v['id']), v.get('params') or {},
Portfolio.load(pp, tp)))
rows_data.sort(key=lambda c: c['summary'].get('total_return_pct', 0) or 0, reverse=True)
scanned = _load(config.LAST_SCAN_PATH, {}).get('scanned_at', '')[:16].replace('T', ' ')
head = (f'
실시간 누적 비교 · 마지막 스캔 {scanned or "—"} '
f'· {len(rows_data)}개 (메인 포함)
'
'
⚠ 같은 라이브 시세에 서로 다른 튜닝 동시 운영. '
'기간 길수록·거래 많을수록 신뢰 · 변이는 백테스트 탭의 「+ 비교군에 추가」로 등록
')
rows = ''
for i, v in enumerate(rows_data):
s = v['summary']
ret = s.get('total_return_pct', 0) or 0
rc = '#22c55e' if ret >= 0 else '#ef4444'
is_main = v['id'] == 'main'
tag = ' 메인' if is_main else ''
plabel = '·'.join(f'{k.split("_")[0]}{val:g}' for k, val in (v.get('params') or {}).items()) or '기본값'
actions = '' if is_main else (
f'
'
f''
f''
f'
'
)
rows += (
f''
f'#{i+1}{_esc(v["name"])}{tag}'
f'{ret:+g}%'
f'
'
)
return head + rows
def render_html() -> str:
snap = _load(config.LAST_SCAN_PATH, {})
pf = Portfolio.load() # 계좌 진실원 — 스캔 스냅샷 없어도 현재 상태 표시
trades = []
if config.TRADES_PATH.exists():
for line in config.TRADES_PATH.read_text().splitlines()[-30:]:
try:
trades.append(json.loads(line))
except Exception:
pass
trades.reverse()
s = pf.summary()
scanned = snap.get('scanned_at', '')[:19].replace('T', ' ')
ret = s.get('total_return_pct', 0) or 0
ret_color = '#22c55e' if ret >= 0 else '#ef4444'
# ---- KPI ----
kpis = (
_kpi('평가자산', f"{s.get('equity', 0):,}원", f"초기 {s.get('initial', 0):,}") +
_kpi('총보유(주식)', f"{s.get('held_value', 0):,}원") +
_kpi('예수금', f"{s.get('cash', 0):,}원") +
f'
수익률
'
f'
{ret:+.2f}%
' +
_kpi('실현손익', f"{s.get('realized_pnl', 0):,}") +
_kpi('승률', f"{s.get('win_rate_pct')}%" if s.get('win_rate_pct') is not None else '—',
f"{s.get('closed_trades', 0)}건 청산") +
_kpi('MDD', f"{s.get('max_drawdown_pct', 0)}%") +
_kpi('보유', f"{s.get('open_positions', 0)}종목")
)
# 기준지수 대비(알파) — 시작일 대비 같은 기간 지수 매수후보유와 비교
bench = snap.get('benchmark', {}) or {}
for idx, lbl in (('KOSPI', '코스피'), ('KOSDAQ', '코스닥')):
b = (bench.get('indices') or {}).get(idx, {})
a, ip = b.get('alpha'), b.get('pct')
if a is None:
kpis += _kpi(f'{lbl} 대비', '—', '지수 데이터 없음')
else:
ac = '#22c55e' if a >= 0 else '#ef4444'
kpis += (f'
{lbl} 대비(알파)
'
f'
{a:+.2f}%p
'
f'
지수 {ip:+.2f}%
')
# ---- 보유 포지션 ----
pos_rows = ''
for p in pf.positions.values():
entry, qty = p['entry_price'], p['qty']
cur = p.get('cur_price', entry)
pnl = (cur / entry - 1) * 100
pnl_amt = int(round((cur - entry) * qty))
pc = '#22c55e' if pnl >= 0 else '#ef4444'
evalv = cur * qty
trail = ' 트레일링' if p.get('trailing_on') else ''
tr_n, tr_max = p.get('tranches', 1), config.ENTRY_TRANCHES
scaled = ' 일부익절' if p.get('scaled_out') else ''
entry_at = _esc((p.get('entry_at') or '')[:16].replace('T', ' '))
pos_rows += (
f''
f'{_esc(p["name"])}'
f'{cur:,} {pnl:+.1f}%'
f'
'
f'
수량{qty:,}주 ({tr_n}/{tr_max}분할){scaled}
'
f'
평단{entry:,.0f}원
'
f'
평가금액{evalv:,}원
'
f'
평가손익{pnl_amt:+,}원 ({pnl:+.1f}%)
'
f'
손절{p["stop"]:,}원
'
f'
목표{p["target"]:,}원{trail}
'
f'
진입사유{_esc(p.get("entry_reason",""))}
'
f'
진입시각{entry_at}
'
f'
'
)
pos_table = pos_rows or '
보유 종목 없음
'
# ---- 판단표 (state 그룹) ----
decisions = snap.get('decisions', [])
by_state = {}
for d in decisions:
by_state.setdefault(d.get('state', 'NODATA'), []).append(d)
judge_html = ''
for st in STATE_ORDER:
items = by_state.get(st, [])
if not items:
continue
label, color = STATE_BADGE.get(st, (st, '#888'))
judge_html += f'
{label} {len(items)}
'
for d in items:
chips = _chips(d.get('checks', []))
src = '/'.join(ORIGIN_LABEL.get(s, s) for s in d.get('sources', []))
price = d.get('price')
price_h = f'{price:,}' if isinstance(price, (int, float)) else '—'
pnl_h = f' · {d["pnl_pct"]:+.1f}%' if d.get('pnl_pct') is not None else ''
del_btn = (f'')
judge_html += (
f'
'
)
# 관찰목록 중 아직 스캔 안 된(방금 추가 등) 종목 → '스캔 대기'로 즉시 표시
from . import universe as umod
wl = umod._load_watchlist()
decided = {d.get('code') for d in decisions}
pending = [(c, e) for c, e in wl.items() if c not in decided]
if pending:
judge_html += '
스캔 대기 {}
'.format(len(pending))
for code, e in pending:
src = ORIGIN_LABEL.get(e.get('origin', ''), e.get('origin', ''))
del_btn = (f'')
judge_html += (
f'
'
f'{_esc(e.get("name") or code)}{_esc(code)} · {_esc(src)}'
f'다음 스캔 대기{del_btn}
'
)
mkt = snap.get('market_ok', {})
mkt_h = ' · '.join(f'{k} {"양호" if v else "약세"}' for k, v in mkt.items()) or '—'
market_html = _market_panel()
tune_html = _tune_panel()
bt_html = _backtest_panel()
cmp_html = _compare_panel()
# ---- 거래내역 ----
tr_rows = ''
for r in trades:
ts = _esc(r["ts"][5:16].replace('T', ' '))
if r['side'] == 'BUY':
st_line = (f'
손절/목표'
f'{r["stop"]:,} / {r["target"]:,}
') if r.get('stop') else ''
tr_rows += (
f''
f'매수{_esc(r["name"])}'
f'{ts}'
f'
'
f'
체결{r["qty"]}주 @ {r["price"]:,}원
'
f'
매수금액{r.get("cost",0):,}원
'
f'
사유{_esc(r.get("reason",""))}
'
f'{st_line}'
+ (f'
판단근거
{_chips(r.get("signals"))}
'
if r.get('signals') else '')
+ '
'
)
else:
pc = '#22c55e' if r.get('pnl', 0) >= 0 else '#ef4444'
tr_rows += (
f''
f'매도{_esc(r["name"])}'
f'{r.get("pnl_pct",0):+.1f}%{ts}'
f'
종목명 일부만 입력해도 후보가 떠요 · 후보를 누르면 추가 · 자산(비하이브 워치·관심·보유)은 자동 편입 · 삭제는 카드의 ✕ (수동만) · 보유 중이면 삭제해도 청산까지 추적
{judge_html}
시장 동향
{market_html}
파라미터 튜닝
{tune_html}
백테스트 순위표
{bt_html}
병렬 페이퍼 비교
{cmp_html}
60초 자동 새로고침 · 당겨서 새로고침 · sim/ 가상 엔진 (실주문 아님)
{PTR_HTML}
{REFRESH_JS}
{_WATCH_SEARCH_JS}
"""
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
parsed = urllib.parse.urlparse(self.path)
path = parsed.path
if path == '/api/stock_search':
from . import universe as umod
q = (urllib.parse.parse_qs(parsed.query).get('q') or [''])[0]
try:
results = umod.search_stocks(q)
except Exception:
results = []
body = json.dumps({'results': results}, ensure_ascii=False).encode('utf-8')
self.send_response(200)
self.send_header('Content-Type', 'application/json; charset=utf-8')
self.send_header('Content-Length', str(len(body)))
self.end_headers()
self.wfile.write(body)
return
if path not in ('/', '/index.html'):
self.send_response(404)
self.end_headers()
return
body = render_html().encode('utf-8')
self.send_response(200)
self.send_header('Content-Type', 'text/html; charset=utf-8')
self.send_header('Content-Length', str(len(body)))
self.end_headers()
self.wfile.write(body)
def do_POST(self):
if self.path == '/run_sweep':
_trigger_sweep()
self.send_response(303)
self.send_header('Location', '/?t=lab&s=bt')
self.end_headers()
return
if self.path in ('/variants_add', '/variants_reset_one', '/variants_remove'):
self._handle_variants()
return
if self.path in ('/watch_add', '/watch_remove'):
self._handle_watch()
return
if self.path != '/save_params':
self.send_response(404)
self.end_headers()
return
length = int(self.headers.get('Content-Length') or 0)
raw = self.rfile.read(length).decode('utf-8') if length else ''
form = {k: v[-1] for k, v in urllib.parse.parse_qs(raw).items()}
try:
if form.get('reset'):
config.reset_params()
else:
config.save_params(form)
except Exception as e:
sys.stderr.write(f'[sim_web] save_params 실패: {e}\n')
self.send_response(303)
self.send_header('Location', '/?t=lab&s=tune&saved=1')
self.end_headers()
def _handle_variants(self):
"""비교군 관리 — 추가(백테스트 행)·성적 초기화·삭제(비교 행). 파일 IO만이라 즉시 완료."""
from . import variants as vmod
length = int(self.headers.get('Content-Length') or 0)
raw = self.rfile.read(length).decode('utf-8') if length else ''
form = {k: v[-1] for k, v in urllib.parse.parse_qs(raw).items()}
# 어느 탭에서 눌렀는지 (back=bt → 백테스트 탭 유지, 없으면 비교 탭). add는 항상 bt.
back = form.get('back')
dest = f'/?t=lab&s={back}' if back in ('bt', 'cmp') else '/?t=lab&s=cmp'
try:
if self.path == '/variants_add':
keys = [spec['key'] for spec in config.TUNABLE]
params = {k: _num(form[k]) for k in keys if k in form}
main_p = config.load_params()
is_main = params and _pkey(params) == _pkey({k: main_p[k] for k in params if k in main_p})
if params and not is_main: # 메인과 동일 조합은 중복이라 추가 X
vmod.add_variant(params)
dest = '/?t=lab&s=bt' # 백테스트 탭에서 연속 추가 가능하도록 유지
elif self.path == '/variants_reset_one' and form.get('id'):
vmod.reset_variant(form['id'])
elif self.path == '/variants_remove' and form.get('id'):
vmod.remove_variant(form['id'])
except Exception as e:
sys.stderr.write(f'[sim_web] variants {self.path} 실패: {e}\n')
self.send_response(303)
self.send_header('Location', dest)
self.end_headers()
def _handle_watch(self):
"""관찰목록 관리 — 수동 추가(/watch_add)·삭제(/watch_remove). 결과를 쿼리로 전달해 배너 피드백."""
from . import universe as umod
length = int(self.headers.get('Content-Length') or 0)
raw = self.rfile.read(length).decode('utf-8') if length else ''
form = {k: v[-1] for k, v in urllib.parse.parse_qs(raw).items()}
dest = '/?t=watch'
try:
if self.path == '/watch_add' and (form.get('q') or '').strip():
q = form['q'].strip()
before = set(umod._load_watchlist())
res = umod.add_manual(q)
if res is None:
dest += '&werr=' + urllib.parse.quote(q) # 종목 못 찾음
elif res['code'] in before:
dest += '&wdup=' + urllib.parse.quote(res.get('name') or res['code']) # 이미 있음
else:
dest += '&wadd=' + urllib.parse.quote(res.get('name') or res['code']) # 추가됨
elif self.path == '/watch_remove' and form.get('code'):
code = form['code']
nm = umod._load_watchlist().get(code, {}).get('name') or code
umod.remove(code)
dest += '&wrem=' + urllib.parse.quote(nm)
except Exception as e:
sys.stderr.write(f'[sim_web] watch {self.path} 실패: {e}\n')
dest += '&werr=' + urllib.parse.quote('처리 오류')
self.send_response(303)
self.send_header('Location', dest)
self.end_headers()
def log_message(self, *a):
pass
def main():
cmd = sys.argv[1] if len(sys.argv) > 1 else 'serve'
if cmd == 'render':
sys.stdout.write(render_html())
return
server = ThreadingHTTPServer((BIND_HOST, BIND_PORT), Handler)
print(f'sim dashboard on http://{BIND_HOST or "0.0.0.0"}:{BIND_PORT} @ {datetime.now(config.KST).isoformat()}',
flush=True)
server.serve_forever()
if __name__ == '__main__':
main()