"""시뮬 대시보드 — 별도 읽기전용 웹서버 (포트 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 hashlib 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, engine from .portfolio import Portfolio BIND_HOST = '' BIND_PORT = 18792 STATE_BADGE = { 'SELL': ('매도', '#ef4444'), 'ADD': ('추격매수', '#0f7a34'), '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=) + 저장 배너 + 자동새로고침(60s) — 펼침(details)·스크롤을 sessionStorage로 보존해 # 전체 리로드여도 보던 화면이 유지되게 한다 (자동·당겨서·수동 새로고침 공통). 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) def _vid_disp(vid) -> str: """고유번호 표시용 — 'v37' → '#37' (내부 id·URL은 raw 유지).""" v = str(vid or '') return '#' + v[1:] if v[:1] == 'v' and v[1:].isdigit() else v _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', 'RSI_OVERBOUGHT': 'RSI↑', 'RSI_OVERSOLD': 'RSI↓', 'MAX_HOLD_DAYS': 'HOLD', 'RISK_PER_TRADE_PCT': 'RISK'} def _plabel(params) -> str: return '·'.join(f'{_PSHORT.get(k, k)}{v:g}' for k, v in (params or {}).items()) # 평가 항목 설명 — 관심종목/판단근거 칩의 ⓘ 버튼 클릭 시 토스트로 표시 (쉬운 말로) _CHK_HELP = { '추세': {'t': '추세 (20일선 위)', 'd': '현재가가 최근 20일 평균값(20일 이동평균선)보다 위에 있는지 봐요. 위에 있으면 최근 한 달 흐름이 상승 쪽이라는 뜻이에요. 자동매매는 떨어지는 칼을 잡지 않으려고, 추세가 위로 향한 종목만 매수 후보로 봅니다.'}, '정배열': {'t': '정배열 (5일선 > 20일선)', 'd': '단기 평균(5일선)이 장기 평균(20일선)보다 위에 있는 상태예요. 최근 5일 기세가 한 달 평균보다 강하다 = 상승에 가속이 붙은 모양이에요. 정배열이면 추세가 살아있다고 봅니다.'}, '수급': {'t': '수급 (외국인·기관 순매수)', 'd': '최근 5일 동안 외국인과 기관이 이 종목을 순매수(더 많이 사들임)했는지 봐요. 돈이 큰 큰손이 사면 주가가 잘 받쳐지는 경향이 있어요. 외국인·기관 둘 중 하나라도 순매수면 통과예요. (숫자 단위는 천 주)'}, '애널': {'t': '애널리스트 의견', 'd': '증권사 애널리스트들의 종합 의견이에요. 보통 5점 만점에 높을수록 매수 추천이에요. ETF처럼 분석 대상이 아닌 종목은 데이터가 없어서 그냥 통과 처리해요.'}, '투자의견': {'t': '투자의견 (5점 만점)', 'd': '증권사 애널리스트들의 평균 투자의견 점수예요. 5점에 가까울수록 "사라"는 의견이 많다는 뜻이고, 보통 4점 이상이면 전문가들이 긍정적으로 본다고 판단해요.'}, '상승여력': {'t': '상승여력 (목표가 대비)', 'd': '애널리스트 평균 목표주가가 지금 가격보다 몇 % 높은지예요. +20%면 전문가들이 20% 더 오를 여지를 본다는 의미예요. 이미 목표가 근처라 여력이 적으면 매력이 줄어요.'}, '목표가리비전': {'t': '목표가 리비전 (상향/하향)', 'd': '최근 애널리스트들이 목표주가를 올리는 중인지 내리는 중인지 봐요. "상향"이면 회사 전망이 좋아지는 방향이라 긍정 신호예요. "하향"이면 반대고요.'}, '어닝서프라이즈': {'t': '어닝 서프라이즈', 'd': '가장 최근 실적이 시장이 예상했던 것(컨센서스)보다 좋았는지(+) 나빴는지(−)예요. 예상보다 잘 나오면(서프라이즈 +) 주가에 긍정적으로 작용하는 경우가 많아요.'}, '시장강세': {'t': '시장 강세 (장 전체 분위기)', 'd': '그날 시장 전체 분위기(상승비율+지수추세+당일등락 조합)예요. 시장이 무너지는 날엔 좋은 종목도 같이 빠지기 쉬워서 참고하는데, 이 항목 하나로 매수가 막히는 건 아니에요 — 전략의 "약세장대처" 모드에 따라 갈려요. 회피=약세면 안 삼 · 안정(기본)=돌파만 삼 · 적극=눌림목도 삼. 그래서 약세(!)인데도 적극 전략은 정상적으로 매수할 수 있어요.'}, '과열': {'t': '과열 아님 (회전율)', 'd': '그날 거래 회전율(거래량을 상장주식 수로 나눈 값)이 평소 평균보다 지나치게(3배 이상) 높지 않은지 봐요. 너무 과열되면 단기 꼭대기일 위험이 커서, 그럴 땐 신규매수를 잠깐 보류해요.'}, '돌파': {'t': '돌파 (전고점)', 'd': '현재가가 최근 고점을 뚫고 올라갔는지예요. 전고점 돌파는 새로운 상승의 출발 신호로 봐요. 거래량·회전율 급증이 함께 나오면 "진짜 돌파"로 판단해 바로 매수합니다.'}, '거래량': {'t': '거래량 급증', 'd': '그날 거래량이 평소 평균보다 크게 많은지 봐요. 거래가 확 늘었다는 건 사람들의 관심이 몰렸다는 뜻이라, 돌파에 힘이 실렸는지 확인하는 신호예요.'}, '회전율': {'t': '회전율 급증', 'd': '회전율(거래량 ÷ 상장주식 수)이 평소보다 크게 늘었는지예요. 거래량 급증과 함께 보면 돌파의 신뢰도를 높여줘요. 데이터가 없으면 그냥 통과 처리해요.'}, '눌림목': {'t': '눌림목 도달', 'd': '강한 종목이 잠깐 쉬며 적정 매수가까지 내려왔는지 봐요. 적정가는 "20일선"과 "전일 종가 − 하루 변동폭(ATR)" 중 더 높은 값이에요. 비싸게 쫓아 사지 않고, 추세는 살아있되 잠깐 눌린 자리에서 싸게 담으려는 지정가 진입 자리예요.'}, 'RSI': {'t': 'RSI (과매수·과매도)', 'd': '최근 오른 폭과 내린 폭을 견줘 0~100으로 나타낸 지표예요. 70~80 위면 단기 과열(너무 많이 올랐다), 20~30 아래면 과매도(너무 많이 빠졌다)로 봐요. 돌파 매수는 과열이면 추격을 피하고, 눌림목 매수는 과매도(아직 떨어지는 중)면 반등을 기다려요.'}, '손절선': {'t': '손절선', 'd': '이 가격 아래로 내려가면 손실을 확정하고 파는 기준선이에요. 진입가에서 변동폭(ATR)이나 최근 스윙 저점을 빼서 정해요. 트레일링이 켜지면 고점을 따라 위로 올라가며 번 수익을 지켜요.'}, } def _chk_key(label) -> str: """체크 항목 라벨 → 설명 키 (라벨에 숫자·괄호가 섞여 있어 접두사로 매칭).""" l = (label or '').strip() for pre, key in (('추세', '추세'), ('정배열', '정배열'), ('수급', '수급'), ('애널', '애널'), ('투자의견', '투자의견'), ('상승여력', '상승여력'), ('목표가', '목표가리비전'), ('어닝', '어닝서프라이즈'), ('시장', '시장강세'), ('과열', '과열'), ('돌파', '돌파'), ('거래량', '거래량'), ('회전율', '회전율'), ('눌림목', '눌림목'), ('RSI', 'RSI'), ('손절', '손절선')): if l.startswith(pre): return key return '' def _chips(checks) -> str: """신호 체크리스트 → 칩 HTML (판단근거 표시 공용). 각 항목에 ⓘ 버튼. 진입경로 칩(돌파 3종 / 눌림목)은 'OR 그룹' 박스로 묶어 표시 — 둘 중 하나만 충족하면 매수라는 구조가 화면에서 보이게. 그룹 라벨의 ✓/✗는 그 경로 전체 충족 여부.""" checks = checks or [] _BRK = ('돌파(전고점)', '거래량 급증', '회전율 급증', '회전율') _PUL = ('눌림목 도달',) def chip(c): k = _chk_key(c.get('label')) info = f'' if k else '' ok = c.get('ok') mark, color = _check_mark(ok), _check_color(ok) # 시장 강세는 그 자체로 매수를 막는 필수조건이 아니라 약세장대처 모드로 분기되는 # 소프트 게이트 — 미충족이어도 빨간 ✗(위반 인상) 대신 노란 ! (참고) if c.get('label') == '시장 강세' and ok is False: mark, color = '!', '#eab308' return ( f'' f'{mark} {_esc(c.get("label"))}' f'{_esc(c.get("detail"))}{info}' ) base = [c for c in checks if c.get('label') not in _BRK + _PUL] brk = [c for c in checks if c.get('label') in _BRK] pul = [c for c in checks if c.get('label') in _PUL] out = ''.join(chip(c) for c in base) if brk or pul: hint = ('매수 진입은 두 경로 중 하나만 충족하면 돼요.\n' '① 돌파: 전고점 돌파 + 거래량·회전율 급증이 모두 충족되면 즉시 매수\n' '② 눌림목: 추세는 살아있는데 적정가까지 내려오면 매수\n' '각 줄 앞의 ✓/✗는 그 경로 전체의 충족 여부예요. 위쪽 칩들은 필수 조건.') def path_line(title, items): oks = [c.get('ok') for c in items] ok = all(v in (True, None) for v in oks) and any(v is True for v in oks) mark, color = ('✓', '#22c55e') if ok else ('✗', '#ef4444') return (f'
' f'{mark} {title}' f'{"".join(chip(c) for c in items)}
') lines = '' if brk: lines += path_line('① 돌파', brk) if pul: lines += path_line('② 눌림목', pul) out += (f'
' f'
' f'진입경로 — 둘 중 하나만 충족하면 매수 ⓘ
{lines}
') return out # 전략 튜닝 탭 파라미터 설명 — 같은 ⓘ 토스트 핸들러 재사용 (data-k = 파라미터 key) _TUNE_HELP = { 'SMA_SHORT': {'t': '단기 이동평균 (일)', 'd': '단기 추세를 보는 평균선 기간이에요. 5면 최근 5일 평균이에요. 짧을수록 가격 변화에 더 민감하게(빠르게) 반응해요. "정배열(단기선>장기선)" 판단에 쓰여요.'}, 'SMA_LONG': {'t': '장기 이동평균 (일)', 'd': '큰 추세를 보는 평균선 기간이에요. 20이면 최근 20일(약 한 달) 평균이에요. "추세(이 선 위인지)"와 "정배열"의 기준선이에요. 길수록 큰 흐름만 봐요.'}, 'ATR_PERIOD': {'t': 'ATR 기간 (일)', 'd': 'ATR은 하루에 가격이 보통 얼마나 출렁이는지(평균 변동폭)예요. 이 기간만큼의 변동폭을 평균내요. 손절·눌림목·트레일링 폭을 계산하는 데 쓰여요.'}, 'BREAKOUT_LOOKBACK': {'t': '돌파 고점 구간 (일)', 'd': '"전고점 돌파"를 따질 때 며칠간의 최고가를 기준으로 할지예요. 20이면 최근 20일 고점을 넘어야 돌파로 봐요. 길수록 더 의미 있는 고점을 요구해요.'}, 'SWING_LOW_LOOKBACK': {'t': '스윙 저점 구간 (일)', 'd': '손절선을 잡을 때 참고하는 최근 저점 구간이에요. 이 기간의 저점 근처를 손절 후보로 써서, 의미 있는 지지선 아래로 빠지면 정리해요.'}, 'VOLUME_AVG_PERIOD': {'t': '거래량 평균 구간 (일)', 'd': '"거래량 급증"을 판단할 때 비교 기준이 되는 평소 평균 거래량을 며칠로 낼지예요. 이 평균보다 확 늘면 관심이 몰린 걸로 봐요.'}, 'FLOW_DAYS': {'t': '수급 합산 (일)', 'd': '외국인·기관 순매수를 며칠치 합쳐서 볼지예요. 5면 최근 5일 합계예요. 길수록 하루 변동이 아닌 꾸준한 큰손 매수를 봐요.'}, 'VOLUME_BREAKOUT_MULT': {'t': '돌파 거래량 배수', 'd': '돌파로 인정하려면 그날 거래량이 평소 평균의 몇 배 이상이어야 하는지예요. 2면 평소 2배 넘게 거래돼야 "진짜 돌파"로 봐요. 높일수록 조건이 까다로워져요.'}, 'TURNOVER_BREAKOUT_MULT': {'t': '돌파 회전율 배수', 'd': '돌파 인정에 필요한 회전율(거래 활발도) 배수예요. 거래량 배수와 함께 봐서 돌파에 힘이 실렸는지 확인해요.'}, 'TURNOVER_OVERHEAT_MULT': {'t': '과열 보류 회전율 배수', 'd': '회전율이 평소의 몇 배를 넘으면 "과열"로 보고 신규매수를 보류할지예요. 3이면 평소 3배 넘게 과열되면 단기 꼭대기 위험으로 잠깐 쉬어요.'}, 'PULLBACK_ATR_MULT': {'t': '눌림목 ATR 배수', 'd': '눌림목 매수가를 "전일 종가 − (변동폭 ATR × 이 배수)"로 잡아요. 클수록 더 깊이 떨어진 자리에서 사려고 해요 — 더 싸게 사지만 기회는 줄어들어요.'}, 'ENTRY_TRANCHES': {'t': '분할매수 횟수', 'd': '한 종목 목표 비중을 몇 번에 나눠 살지예요. 3이면 1/3씩 세 번에 걸쳐 들어가요. 한 번에 다 사지 않아 평균 단가 위험을 줄여요.'}, 'ADD_ATR_MULT': {'t': '추격매수 ATR 배수', 'd': '추가 매수(추격)할 때, 직전 매수가보다 "변동폭 ATR × 이 배수"만큼 더 오르면 다음 분할을 사요. 추세가 계속될 때만 따라 올라타는 기준이에요.'}, 'RSI_OVERBOUGHT': {'t': 'RSI 과열 컷 (돌파)', 'd': '돌파 매수할 때 RSI가 이 값 이상이면 단기 과열로 보고 추격을 보류해요. 78이면 RSI 78 이상일 때 안 사요. 낮출수록 더 깐깐하게(덜) 사고, 너무 높이면 과열 종목도 사게 돼요.'}, 'RSI_OVERSOLD': {'t': 'RSI 과매도 컷 (눌림목)', 'd': '눌림목 매수할 때 RSI가 이 값 미만이면 아직 떨어지는 중(과매도)으로 보고 진입을 미뤄요. 25면 RSI 25 미만일 때 기다려요. "떨어지는 칼을 잡지 않으려는" 안전장치예요.'}, 'STOP_ATR_MULT': {'t': '손절 ATR 배수', 'd': '손절선을 "매수가 − (변동폭 ATR × 이 배수)"로 잡아요. 2면 하루 변동폭의 2배까지 빠지면 손절이에요. 클수록 여유롭게 두지만(잘 안 잘림) 손실은 커질 수 있어요.'}, 'RR_RATIO': {'t': '손익비 (목표 ÷ 손절)', 'd': '목표 수익이 손절 위험의 몇 배인지예요. 2면 1 잃을 위험에 2 벌 목표를 둬요. 높일수록 큰 수익을 노리지만 목표 도달은 어려워져요.'}, 'SCALE_OUT_FRAC': {'t': '분할매도 익절 비율', 'd': '목표가에 닿으면 보유 수량 중 몇 %를 먼저 익절할지예요. 0.5면 절반을 팔아 수익을 챙기고, 나머지는 더 끌고 가며 트레일링으로 관리해요.'}, 'TRAIL_ATR_MULT': {'t': '트레일링 ATR 배수', 'd': '익절 후 남은 물량의 "따라가는 손절선"을 고점에서 "변동폭 ATR × 이 배수"만큼 아래에 둬요. 주가가 오르면 같이 올라가고, 떨어지면 그 선에서 정리해요.'}, 'TRAIL_ATR_MULT_TIGHT': {'t': '트레일링 강화 배수', 'd': '수익이 충분히 났을 때 트레일링 폭을 더 좁게(타이트하게) 조이는 배수예요. 번 것을 더 꼭 지키려고 고점에 더 가깝게 따라붙어요.'}, 'MAX_HOLD_DAYS': {'t': '시간손절 (보유 최대 일)', 'd': '한 종목을 산 뒤 이 일수(달력 기준)가 지나도 진전이 없으면(트레일링 시작 전이고 진입가도 못 넘김) 정리해서 자본을 다른 기회로 돌려요. 20이면 약 한 달이에요. 0이면 시간손절을 끕니다.'}, 'ANALYST_MIN_UPSIDE_PCT': {'t': '최소 상승여력 (%)', 'd': '매수 후보가 되려면 애널리스트 목표가까지 최소 몇 %의 여력이 있어야 하는지예요. 이보다 낮으면(이미 목표가 근처면) 매력이 부족하다고 봐요.'}, 'ANALYST_SELL_OPINION_MAX': {'t': '매도의견 컷 (1~5)', 'd': '애널리스트 평균 의견 점수가 이 값 이하로 낮으면(매도에 가까우면) 매수 후보에서 빼요. 전문가가 부정적인 종목을 거르는 방향성 안전장치예요.'}, 'MARKET_BREADTH_MIN': {'t': '시장 상승비율 하한', 'd': '그날 오른 종목 비율이 이 값 미만이면 "약세장"으로 봐요. 0.15면 15%예요. 약세장에선 바닥잡기 매수는 미루고 강한 돌파만 허용해요.'}, 'BEAR_ENTRY_MODE': {'t': '약세장 매수 모드', 'd': '시장이 약세 국면(혼조·약세·급락)일 때 어떻게 살지예요. 0=완전 현금화(돌파도 안 사고 무조건 쉼) · 1=돌파만 허용(강하게 치고 오르는 종목만, 바닥잡기는 안 함) · 2=눌림목도 허용(약세장에도 적정가 오면 저가 매수). 숫자가 클수록 약세장에 적극적이에요. 기본 1.'}, 'MAX_POSITIONS': {'t': '동시 보유 최대 종목', 'd': '한 번에 들고 갈 수 있는 최대 종목 수예요. 분산은 되지만 너무 많으면 관리가 어려워요. 이 수에 차면 신규매수를 멈춰요. "종목당 균등비중 = 자산 ÷ 이 수"의 기준이기도 해요.'}, 'RISK_PER_TRADE_PCT': {'t': '종목당 리스크 비율', 'd': '한 종목에서 감수할 손실(손절까지 갔을 때)을 자산의 몇 %로 제한할지예요. 0.01이면 1%. 수량 = 자산 × 이 비율 ÷ 손절폭으로 정해서, 손절폭이 큰(변동성 큰) 종목은 자동으로 적게 사요. 종목마다 위험을 균등하게 맞춰 큰 손실 한 방을 막아요. 0이면 끄고 균등비중으로 사요.'}, } _CHK_HELP_JS = (""" """).replace('__HELP_JSON__', json.dumps({**_CHK_HELP, **_TUNE_HELP}, ensure_ascii=False)) def _kpi(label, value, sub=''): sub_h = f'
{_esc(sub)}
' if sub else '' return f'
{_esc(label)}
{_esc(value)}
{sub_h}
' FAV_PATH = config.STATE_DIR / 'favorites.json' # 관심 전략 마킹 아이콘 — 하나당 전략 1개만 (다른 전략에 쓰면 옮겨짐). 순서 = 우선순위(관심종목 탭 기본 시각). FAV_ICONS = ['⭐', '🔥', '💎', '🎯', '👀', '🚀'] def _load_favs() -> dict: """{아이콘: vid} — 아이콘 하나당 전략 1개. 구버전(list)은 순서대로 마이그레이션.""" try: d = json.loads(FAV_PATH.read_text()) except Exception: return {} if isinstance(d, list): return {ic: v for ic, v in zip(FAV_ICONS, d)} return {ic: v for ic, v in d.items() if ic in FAV_ICONS} def _save_favs(favs: dict): config.STATE_DIR.mkdir(parents=True, exist_ok=True) FAV_PATH.write_text(json.dumps(favs, ensure_ascii=False)) def _kvline(k, v): """한 줄 항목(라벨 좌 / 값 우). v는 신뢰 HTML(색상·muted 등 허용).""" return f'
{k}{v}
' 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 _mkt_cache = {'data': None, 'ts': 0.0} _MKT_TTL = 60.0 def _get_live_market() -> dict | None: """장중 실시간 시장통계(상승하락·ADR·투자자 순매수) — 네이버 m.stock, 60s 캐시. 실패 시 None.""" now = time.time() if _mkt_cache['data'] and now - _mkt_cache['ts'] < _MKT_TTL: return _mkt_cache['data'] try: sys.path.insert(0, str(config.SCRIPTS)) import market_indicators_sync as mis out = {} for sym, lab in (('KOSPI', '코스피'), ('KOSDAQ', '코스닥')): rr = mis.fetch_market(sym, lab) if rr: rr['date'] = rr.get('bizdate') or '' out[sym] = rr if out: _mkt_cache['data'] = out _mkt_cache['ts'] = now return out except Exception as e: sys.stderr.write(f'[sim_web] 실시간 시장통계 fetch 실패: {e}\n') return None 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() live = _get_live_market() # 세션 무관 항상 최신(네이버는 장 마감 후에도 당일 최종값 제공), 실패 시만 EOD 폴백 regimes = engine.market_regime_map() # 6단계 국면 (매수판단과 동일 소스) 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], **live[k]} if live and k in live else 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'
지수' f'{ix["price"]:,.2f} ' f'{ix["change"]:+,.2f} ({ix["pct"]:+.2f}%)
') 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 grade = regimes.get(k, 'mixed') r_em, r_lbl, r_buy, r_desc, r_color = engine.REGIME[grade] vlabel, vcolor = f'{r_em} {r_lbl}', r_color 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'
{label} ' f'{_esc(r.get("date",""))}' f'{vlabel}
' f'{idx_row}' f'
ADR{adr_s}
' f'
상승 / 하락{rise:,} / {fall:,} (보합 {steady:,})
' f'
상한 / 하한{r.get("upper",0):,} / {r.get("lower",0):,}
' f'
상승비율{breadth:.1f}%
' f'
국면{vlabel}
' f'
{r_desc}{" · 신규매수 허용" if r_buy else ""}
' f'
개인 순매수{_signed(r.get("personal"))}
' f'
외국인 순매수{_signed(r.get("foreign"))}
' f'
기관 순매수{_signed(r.get("institutional"))}
' f'
최근 ADR(5일){recent_s}
' f'
' ) if live: live_lbl = '장중 실시간' if engine.is_market_session() else '직전 장 마감 기준' note = f'
{live_lbl}(네이버 m.stock 실시간) · 순매수 단위 억원
' else: note = ('
EOD 폴백(실시간 조회 실패) · 순매수 단위 억원
') 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 = '
' for gname, specs in groups.items(): body += f'

{_esc(gname)}

' for spec in specs: k = spec['key'] body += ( f'
' f'' f'
' ) 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 _market_fit(p: dict) -> tuple[str, str]: """조합(설정값 세트)이 어떤 시장에서 강세/약세인지 — 파라미터 성격 기반 결정론적 설명.""" sma = p.get('SMA_LONG', 20); rr = p.get('RR_RATIO', 2.0); stop = p.get('STOP_ATR_MULT', 2.0) vol = p.get('VOLUME_BREAKOUT_MULT', 1.5); pull = p.get('PULLBACK_ATR_MULT', 1.0) risk = p.get('RISK_PER_TRADE_PCT', 0.01); hold = p.get('MAX_HOLD_DAYS', 0) tags, strong, weak = [], [], [] tags.append('크게 먹으려 오래 들고가는 편' if rr >= 2.5 else '조금씩 빨리 먹고 빠지는 편' if rr <= 1.5 else '적당히 균형 잡힌 편') if sma <= 15: tags.append('시세에 빠르게 반응') elif sma >= 40: tags.append('큰 흐름만 천천히') if risk >= 0.02: tags.append('한 종목에 크게 베팅') elif risk <= 0.005: tags.append('조심조심 적게 베팅') if vol <= 1.3: strong.append('거래가 활발하고 쭉쭉 오르는 장 — 치고 올라갈 때 잘 따라붙어요') weak.append('거래도 뜸하고 옆으로 횡보하는 장 — 오르는 척하다 도로 빠질 때 속아요') elif vol >= 2.0: strong.append('확실하게 크게 터지며 오르는 장') weak.append('조용한 장 — 살 만한 신호가 잘 안 나와서 기회가 적어요') if rr >= 2.5: strong.append('한 방향으로 길게 쭉 가는 장 — 오래 들고 크게 먹어요') weak.append('올랐다 내렸다 왔다갔다 하는 장 — 목표에 닿기 전에 꺾여요') elif rr <= 1.5: strong.append('출렁출렁 자주 움직이는 장 — 조금씩 빨리 먹기 좋아요') weak.append('한 방향으로 크게 가는 장 — 너무 일찍 팔아서 아쉬워요') if stop <= 1.5: strong.append('완만하게 꾸준히 오르는 장 — 손실을 작게 끊어요') weak.append('출렁임이 심한 장 — 자꾸 손절에 걸려서 잘려요') elif stop >= 2.5: strong.append('흔들려도 결국 오르는 장 — 출렁임을 버텨내요') weak.append('확 빠지는 급락장 — 한 번에 크게 손해 봐요') if pull >= 1.5: strong.append('올랐다 잠깐 쉬며 내려주는 장 — 싸게 살 기회가 생겨요') weak.append('쉬지 않고 곧장 오르기만 하는 장 — 살 자리를 안 줘서 못 사요') if risk >= 0.02: strong.append('쭉 오르는 장 — 크게 베팅한 만큼 수익도 커요') weak.append('내리거나 출렁이는 장 — 손실도 그만큼 커서 계좌가 휘청여요') if sma <= 15: weak.append('자잘하게 흔들리기만 하는 장 — 빠른 신호가 자꾸 헛다리 짚어요') elif sma >= 40: weak.append('빠르게 휙휙 바뀌는 장 — 굼떠서 한발 늦게 들어가요') if hold and hold <= 10: strong.append('짧게 끝나는 장 — 돈을 빨리 돌려 다음 기회로 넘어가요') def _dedup(lst): seen, out = set(), [] for x in lst: if x not in seen: seen.add(x); out.append(x) return out strong, weak = _dedup(strong), _dedup(weak) body = ('한마디로: ' + ', '.join(tags) + ' 스타일이에요.\n\n' '💪 이런 장에서 잘 벌어요\n' + '\n'.join('· ' + x for x in strong) + '\n\n⚠️ 이런 장에선 약해요\n' + '\n'.join('· ' + x for x in weak) + '\n\n※ 설정값 성격으로 따져본 거예요 — 참고만 하세요') return '🌐 어떤 시장에 강한가', body def _fit_tokens(p: dict) -> tuple[str, list[str], str]: """조합 성격 토큰 (market, 수식어들, 스타일) — 긴 문장/짧은 나열 공용.""" rr = p.get('RR_RATIO', 2.0); risk = p.get('RISK_PER_TRADE_PCT', 0.01) sma = p.get('SMA_LONG', 20); stop = p.get('STOP_ATR_MULT', 2.0) vol = p.get('VOLUME_BREAKOUT_MULT', 1.5); pull = p.get('PULLBACK_ATR_MULT', 1.0) style = '추세추종' if rr >= 2.5 else '단타' if rr <= 1.5 else '균형' adj = [] if sma <= 15: adj.append('빠른') elif sma >= 40: adj.append('느린') if risk >= 0.02: adj.append('공격') elif risk <= 0.005: adj.append('보수') if risk >= 0.02 or vol <= 1.3: market = '상승장' elif rr >= 2.5: market = '추세장' elif pull >= 1.5: market = '조정장' elif rr <= 1.5 or stop <= 1.5: market = '출렁이는 장' elif stop >= 2.5: market = '변동성 장' else: market = '보통 장' return market, adj, style def _market_fit_summary(p: dict) -> str: """조합 성격을 한 줄로 — 백테스트 탭 행 제목·유형 라벨용. 예: '상승장에 강한 빠른·공격 단타형'.""" market, adj, style = _fit_tokens(p) style_full = ('·'.join(adj) + ' ' if adj else '') + style + '형' return f'{market}에 강한 {style_full}' def _market_fit_words(p: dict) -> str: """짧은 단어 나열 — 비교탭 행 제목·드롭다운용. 예: '상승장·빠른·공격·추세추종'.""" market, adj, style = _fit_tokens(p) m = {'출렁이는 장': '출렁장', '변동성 장': '변동장', '보통 장': '보통장'}.get(market, market) return '·'.join([m] + adj + [style]) def _cat_labels(p: dict) -> tuple[str, str, str, str]: """전략을 4카테고리 두글자 이름으로 — (추세, 진입, 매도, 베팅).""" p = p or {} sma = p.get('SMA_LONG', 20) trend = {10: '민감', 20: '표준', 40: '둔감', 60: '장기'}.get(sma, '표준') entry = '적극' if p.get('VOLUME_BREAKOUT_MULT', 1.5) <= 1.5 else '신중' rr = p.get('RR_RATIO', 2.0) sell = '한방' if rr >= 2.5 else '단타' if rr <= 1.5 else '균형' bet = '집중' if p.get('RISK_PER_TRADE_PCT', 0.01) >= 0.02 else '분산' return trend, entry, sell, bet def _cat_name(p: dict) -> str: """4카테고리 이름 한 줄 — 예 '표준·적극·균형·분산'.""" return '·'.join(_cat_labels(p)) def _cat_detail_html(p: dict) -> str: """펼침용 — 4카테고리 각각의 선택값·쉬운 설명.""" p = p or {} trend, entry, sell, bet = _cat_labels(p) sma = p.get('SMA_LONG', 20); vol = p.get('VOLUME_BREAKOUT_MULT', 1.5) rr = p.get('RR_RATIO', 2.0); risk = p.get('RISK_PER_TRADE_PCT', 0.01) stop = p.get('STOP_ATR_MULT', 2.0); hold = p.get('MAX_HOLD_DAYS', 0) rows = [ ('🟦 추세', trend, f'{sma:g}일선 기준', '추세 — 며칠짜리 평균선으로 보나', f'추세를 며칠짜리 평균선으로 보는지예요. 이 전략은 {sma:g}일선 기준({trend}).\n\n' '· 민감(10일선): 빨리 반응 — 기회 빨리 잡지만 속임수(휩쏘)도 많음\n' '· 표준(20일선): 무난한 중간\n· 둔감(40일선): 천천히 — 늦지만 안정적\n· 장기(60일선): 큰 흐름만'), ('🟩 진입', entry, f'돌파 거래량 {vol:g}배', '진입 — 얼마나 쉽게 사나', f'돌파로 살 때 거래량이 평소의 몇 배는 터져야 사는지예요. 이 전략은 {vol:g}배({entry}).\n\n' '· 적극(1.3배): 문턱 낮음 — 자주 삼\n· 신중(2.0배): 확실히 터질 때만 삼'), ('🟥 매도', sell, f'손익비 {rr:g} · 손절 ATR{stop:g}배' + (f' · 시간손절 {hold}일' if hold else ''), '매도 — 얼마나 길게 먹나', f'목표 수익을 손절폭의 몇 배로 잡는지예요. 이 전략은 손익비 {rr:g}({sell}).\n\n' '· 한방(3.0): 길게 들고 크게 먹기\n· 균형(2.0): 중간\n· 단타(1.5): 짧게 자주 먹기\n\n' f'손절은 변동성(ATR)의 {stop:g}배 빠지면 실행' + (f', {hold}일 지나도 안 오르면 정리해요.' if hold else '해요.')), ('🟨 베팅', bet, f'종목당 리스크 {risk * 100:g}%', '베팅 — 한 종목에 얼마나', f'한 종목에 자산의 몇 %까지 거는지예요. 이 전략은 {risk * 100:g}%({bet}).\n\n' '· 집중(2%): 크게 베팅 — 수익·손실 둘 다 큼\n· 분산(1%): 작게 — 안정적'), ] out = '' for cat, name, val, title, desc in rows: out += (f'
{cat}' f'{_esc(name)} — {_esc(val)}' f'
') return out # 약세장 모드 단어 칩 — 비교군 행·드롭다운에서 한눈에 구분 (긴 접미사 대신). _BEAR_WORD = {0: '회피', 1: '안정', 2: '적극'} _BEAR_NAME = {0: '회피 (현금화)', 1: '안정 (돌파만·기본)', 2: '적극 (눌림허용)'} _BEAR_DESC = { 0: '이 전략은 시장이 약세 국면(혼조·약세·급락)이면 아무것도 안 사고 현금을 지켜요.', 1: '이 전략은 시장이 약세 국면이어도 강하게 치고 오르는 돌파 종목만 사요. 바닥잡기(눌림목 매수)는 안 해요.', 2: '이 전략은 시장이 약세 국면이어도 눌림목 저가매수까지 해요. 셋 중 가장 적극적이에요.', } _BEAR_LEGEND = '\n\n회피=약세장 매수 안 함 · 안정=돌파만 매수(기본) · 적극=눌림목 저가매수도 함' def _bear_mode(p) -> int: """변이 params의 약세장 모드 — 없으면 기본 1(돌파만).""" try: return int((p or {}).get('BEAR_ENTRY_MODE', 1)) except Exception: return 1 def _bear_chip_html(p) -> str: """약세모드 단어 칩 (클릭 시 설명 토스트).""" m = _bear_mode(p) return (f'{_BEAR_WORD[m]}') def _gate_info(p): """진입 게이트 차원 — (이름, 설명). 유형·약세장대처와 독립된 3번째 축.""" p = p or {} off = [] if not int(p.get('TREND_SLOPE_GATE', 1)): off.append('장기선 우상향 요구') if not int(p.get('MID_ARRAY_GATE', 1)): off.append('중기 정배열(5>60) 요구') if not off: return ('보수 (현행)', '추세 조건을 모두 요구해요 — 현재가>장기선·정배열·장기선 우상향·중기 정배열(5>60). ' '안전하지만 매수 기회가 적어요(보수적).') return ('완화 — ' + '·'.join('장기선 기울기' if '우상향' in o else '중기 정배열' for o in off) + ' 면제', '추세 게이트를 완화한 변이예요. 꺼진 조건:\n· ' + '\n· '.join(off) + '\n\n기본 조건(현재가>장기선·정배열)은 그대로라, 단기 반등 초입 종목도 잡아 더 적극적으로 매수해요. ' '보수형(현행)과 라이브로 비교 중입니다.') def _relax_chip_html(p) -> str: """추세 게이트 완화 칩 — 장기선 기울기/중기정배열 게이트를 끈 변이 표시. 현행(둘다 ON)이면 빈 문자열.""" p = p or {} off = [] if not int(p.get('TREND_SLOPE_GATE', 1)): off.append('장기선 우상향 요구') if not int(p.get('MID_ARRAY_GATE', 1)): off.append('중기 정배열(5>60) 요구') if not off: return '' desc = ('추세 게이트 완화 변이 — 다음 조건을 꺼서 더 적극적으로 매수해요:\n· ' + '\n· '.join(off) + '\n\n보수형(현행)과 라이브로 비교 중인 실험 전략입니다.') return (f'완화') def _immediate_chip_html(p) -> str: """즉시매수 칩 — 돌파·눌림목 안 기다리고 조건 통과 시 현재가 즉시 매수하는 변이. 아니면 빈 문자열.""" if not int((p or {}).get('IMMEDIATE_ENTRY', 0)): return '' desc = ('즉시매수 변이 — 추세·정배열·수급·애널을 다 통과한 종목이면, 돌파나 눌림목을 안 기다리고 ' '강세장에서 현재가에 바로 사요. 꾸준히 오르기만 하는 좋은 종목을 놓치지 않으려는 실험 전략이에요. ' '(기다렸다 사는 기존 전략과 라이브 비교 중)') return (f'즉시') def _avg_chip_html(p) -> str: """물타기(애버리징) 칩 — 보유 중 하락 시 손절 대신 추가매수로 평단 낮추는 변이. 아니면 빈 문자열.""" if not int((p or {}).get('AVERAGING_ENTRY', 0)): return '' drop = (p or {}).get('AVERAGING_DROP_PCT', 0.10) desc = ('물타기(애버리징) 변이 — 우량주(추세·정배열 통과)를 산 뒤 직전 매수가보다 ' f'-{drop*100:.0f}% 떨어지면, 손절하지 않고 추가매수해 평균단가를 낮춰요(종목당 10% 한도). ' '단 ①당일 -10%↓ 급락이면 물타기 말고 손절 ②약세장이면 물타기 금지 — 떨어지는 칼·하락장 물타기는 피해요. ' '(손절형 기존 전략과 라이브 비교 중)') return (f'물타기') def _backtest_panel() -> str: """백테스트 결과 — 유형별(세부 라벨) 대표(1위) 조합 요약. 비교군은 이 대표들로 자동 구성되므로 수동 추가/제거 없이 읽기 전용 요약만 보여준다.""" res = _load(config.STATE_DIR / 'backtest_results.json', None) if not res or not res.get('results'): return ('
스윕 미실행 — 터미널에서 python3 -m sim sweep 실행 후 표시됩니다.
') sp = res.get('split', {}) head = (f'
검증 {sp.get("test","?")} · {res.get("count")}개 조합 · ' f'순위=3분할 최악구간 수익(꾸준함) · 다음날 시가 체결+슬리피지 0.2%
') st = _sweep_status() if st.get('state') == 'running': ctrl = ('
' '스윕 계산 중… (약 1~2분 소요 · 끝나면 자동 갱신)
' '') else: err = ('
⚠ 직전 스윕 실패 — 로그 확인 필요
' if st.get('state') == 'error' else '') ctrl = (err + '
' '
') main_p = config.load_params() main_pkey = _pkey({k: main_p[k] for k in res.get('swept_params', []) if k in main_p}) min_tr = res.get('min_trades', 0) or 0 def _sk(r): te = r['test'] enough = (te.get('trades', 0) or 0) >= min_tr worst = r.get('worst_ret') return (1 if enough else 0, worst if worst is not None else -999, te.get('total_return_pct', -999) or -999) def _row_html(r, rank_label): p, te, tr = r['params'], r['test'], r['train'] is_main = _pkey(p) == main_pkey main_badge = ' ' if is_main else '' 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 '' mdd, wr, pf = te.get('mdd_pct'), te.get('win_rate_pct'), te.get('profit_factor') mf_t, mf_b = _market_fit(p) mf_sum = _market_fit_summary(p) wins_row = '' if r.get('windows'): parts = [] for wnd in r['windows']: wret = wnd.get('ret') wc = '#22c55e' if (wret or 0) >= 0 else '#ef4444' parts.append(f'{wret:+g}%' if wret is not None else '—') wins_row = (f'
구간별 검증(3분할)' f'{" / ".join(parts)}
') worst = r.get('worst_ret') worst_h = f'최악 {worst:+g}% · ' if worst is not None else '' return ( f'
{rank_label} ' f'{_esc(mf_sum)}{main_badge}' f'{worst_h}검증 {teret:+g}%' f'
' f'
검증 수익 / 학습 수익' f'{teret:+g}% / {trret:+g}%{of_tag}
' f'{_bt_alpha_row(te)}' f'{wins_row}' f'
검증 MDD{mdd}%
' f'
검증 거래수 / 승률{te.get("trades")}건 · {wr}%
' f'
검증 손익비(PF){pf}
' f'
파라미터{_esc(_plabel(p))}
' f'' f'
' ) # 중복(동일 params) 제거 → 세부 유형(라벨)별 대표(1위)만 추려 요약 uniq = {} for r in res['results']: uniq.setdefault(_pkey(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 # 시장 성격별 그룹 (각 그룹 안은 꾸준함 순) by_market = {} for lbl, r in by_label.items(): by_market.setdefault(lbl.split('에 강한')[0], []).append(r) note = (f'
유형별({len(by_label)}종) 대표 조합 — ' f'각 유형 1위만 요약 · 비교군 반영은 수동(코디에게 요청) (총 고유 {len(uniq)}조합)
' f'※ 약세장 모드(기본/현금화/눌림허용)는 백테스트에 영향이 없어(시장필터 중립) 여기선 안 갈라져요 — ' f'비교군에서만 유형당 3종으로 라이브 검증
') # 유형별 보기 (시장 성격 그룹 — 접이식, 접힌 줄에 그룹 1위 수익 표시) type_rows = '' for mkt in sorted(by_market, key=lambda m: max(_sk(r) for r in by_market[m]), reverse=True): grp = sorted(by_market[mkt], key=_sk, reverse=True) best = grp[0] bret = best['test'].get('total_return_pct', 0) bw = best.get('worst_ret') bc = '#22c55e' if (bret or 0) >= 0 else '#ef4444' bw_h = f'최악 {bw:+g}% · ' if bw is not None else '' rows_html = ''.join(_row_html(r, f'#{j + 1}') for j, r in enumerate(grp)) type_rows += ( f'
' f'{_esc(mkt)}에 강한 유형 ({len(grp)}종)' f'1위 {bw_h}{bret:+g}%' f'
{rows_html}
' ) # 순위별 보기 (전체 36종 통합 꾸준함 순) flat = sorted(by_label.values(), key=_sk, reverse=True) rank_rows = ''.join(_row_html(r, f'#{i + 1}') for i, r in enumerate(flat)) viewbar = ('
' '' '
') toggle_js = """ """ return (head + ctrl + note + viewbar + f'
{type_rows}
' + f'' + toggle_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') # 스캔 시점 '노리는 종목' (variants_compare.json) — id별 매핑 watch_map = {c['id']: c.get('watching', []) for c in (_load(vmod.COMPARE_PATH, {}) or {}).get('variants', [])} # 라이브 거래일수 — equity_history.json 에 거래일마다 (id,ph) 1점씩 → 등장한 거래일 수 (백테스트 제외) _eh = _load(config.STATE_DIR / 'equity_history.json', {}) or {} live_days: dict = {} for _d, _items in _eh.items(): for _e in _items: live_days.setdefault((_e.get('id'), _e.get('ph')), set()).add(_d) def _row(rid, name, params, pf, tpath): s = pf.summary() return {'id': rid, 'name': name, 'params': params, 'summary': s, 'pf': pf, 'tpath': tpath, 'benchmark': benchmark.compare(s.get('total_return_pct'), benchmark.norm_date(pf.created_at), today)} # 변이 계좌 수집 (기준전략 폐지 — 변이만 비교) favs = _load_favs() icon_of = {v: ic for ic, v in favs.items()} # vid → 아이콘 (전략당 1개) # 백테스트 성적 맵 — 각 변이가 어떤 성적으로 등록됐는지 (params 정규화 키 → 순위·최악·검증) bt_map = {} _btres = _load(config.STATE_DIR / 'backtest_results.json', None) or {} _seen = set() _rank = 0 for r in _btres.get('results', []): kk = _pkey(r['params']) if kk in _seen: continue _seen.add(kk) _rank += 1 bt_map[kk] = {'rank': _rank, 'worst': r.get('worst_ret'), 'test': r['test'].get('total_return_pct')} # 백테스트 매칭은 스윕 대상 파라미터만으로 — 약세모드 등 백테스트 중립 파라미터(BEAR_ENTRY_MODE)가 # 변이 params에 붙어 있어도 같은 백테스트 행에 매칭되게 (모드 3종은 백테스트 성적이 동일). _swept = _btres.get('swept_params') or [] def _btkey(p): p = p or {} return _pkey({k: p[k] for k in _swept if k in p} if _swept else p) # 비교군 내부 순위 (1~N) — 등록된 변이들끼리만 백테스트 순위 재매김 (전체 순위는 쌍둥이 제외로 번호가 건너뜀) _vkeys = {_btkey(v.get('params') or {}) for v in defs} # set — 모드 3종이 같은 키를 공유 _border = sorted([k for k in _vkeys if k in bt_map], key=lambda k: bt_map[k]['rank']) bt_grank = {k: i + 1 for i, k in enumerate(_border)} rows_data = [] 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), tp)) # 표시 순서는 고정(메인→변이 등록순) — 새로고침해도 위치 안 바뀜. 순위(#N)만 수익률로 매겨 라벨 표시. _ranked = sorted(rows_data, key=lambda c: c['summary'].get('total_return_pct', 0) or 0, reverse=True) rank_of = {c['id']: n + 1 for n, c in enumerate(_ranked)} scanned = _load(config.LAST_SCAN_PATH, {}).get('scanned_at', '')[:16].replace('T', ' ') head = (f'
실시간 누적 비교 · 마지막 스캔 {scanned or "—"} ' f'· 전략 {len(rows_data)}개
') bt_btn = ('') day_btn = ('') title = (f'

병렬 페이퍼 비교

' f'
{bt_btn}{day_btn}
') 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' fav_ic = icon_of.get(v['id'], '') tag = (f' ') mp = v.get('params') or {} mf_t, mf_b = _market_fit(mp) mf_sum = _market_fit_summary(mp) mf_w = _market_fit_words(mp) bm = _bear_mode(mp) _ph = hashlib.md5(json.dumps(mp, sort_keys=True).encode()).hexdigest()[:8] ndays = len(live_days.get((v['id'], _ph), ())) w = watch_map.get(v['id'], []) _STL = {'BUY': '🟢 매수', 'ADD': '🟢 추격', 'WAIT': '🟡 대기'} watch_html = ''.join( f'
{_STL.get(it.get("state"), it.get("state"))} {_esc(it.get("name"))}' f'{_esc(it.get("reason",""))[:28]}
' for it in w) or '
노리는 종목 없음
' reset_btn = (f'
' f'' f'
') del_btn = (f'
' f'' f'
') det_btn = (f'📈 자세히보기') actions = f'
{det_btn}{reset_btn}{del_btn}
' vid = v['id'] _vk = _btkey(v.get('params') or {}) bt = bt_map.get(_vk) gr = bt_grank.get(_vk) # 평가순 정렬 키 = 고유번호 오름차순 (#1→#108) — 번호 자체가 평가순 부여라 그대로가 가장 직관적. # bt 미매칭 변이만 뒤로. (등수×1000+번호 방식은 #1·#37·#38… 로 보여 뒤섞인 인상 — 2026-06-11 단순화) _vnum = int(vid[1:]) if vid[1:].isdigit() else 999 bt_attr = _vnum if (bt and gr) else 900000 + _vnum if bt and gr: bw = f"{bt['worst']:+g}%" if bt['worst'] is not None else '—' bt_row = (f'
모의평가' f'{gr}위/{len(bt_grank)}
' f'
' f'최악구간 {bw} · 검증 {bt["test"]:+g}%
') else: bt_row = '' rows += ( f'
' f'{rank_of[v["id"]]}위{_bear_chip_html(mp)}{_relax_chip_html(mp)}{_immediate_chip_html(mp)}{_avg_chip_html(mp)} {_vid_disp(vid)} ' f'{_esc(_cat_name(mp))}{tag}' f'🛒{s.get("open_positions", 0)} ' f'{ret:+g}%' f'
' f'
전략 4요소추세·진입·매도·베팅
' f'{_cat_detail_html(mp)}' f'
시장적합{_esc(mf_sum)}' f'
' f'
약세장대처{_BEAR_NAME[bm]}' f'
' f'
운영 기간{ndays or "—"}거래일째 (백테스트 제외)
' f'
평가자산{s.get("equity",0):,}원
' f'
예수금 / 주식{s.get("cash",0):,} / {s.get("held_value",0):,}
' f'
수익률{ret:+g}%
' f'
실현손익{s.get("realized_pnl",0):,}원
' f'
거래 / 승률{s.get("closed_trades",0)}건 · {s.get("win_rate_pct")}%
' f'
MDD / 보유{s.get("max_drawdown_pct",0)}% · {s.get("open_positions",0)}종목
' f'
지수대비(알파){_alpha_inline(v.get("benchmark") or {})}
' f'{bt_row}' f'
📦 보유종목 ({s.get("open_positions",0)})' f'
{_pos_table(v["pf"])}
' f'
📋 거래내역 ({_trade_count(v["tpath"])}건 · {s.get("closed_trades",0)}청산)' f'
{_trades_table(_read_trades(v["tpath"]))}
' f'
🎯 관심종목 ({len(w)})' f'
{watch_html}
' f'{actions}' f'
' ) sortbar = ('
' '' '' '' '
') # 각 버튼: 한 번 누를 때마다 오름/내림 토글. 기본 방향 — 손익/보유/거래=내림(desc), 평가(백테스트순위)=오름(asc) sort_js = """ """ fav_js = (""" """ .replace('__FAVS__', json.dumps(favs, ensure_ascii=False)) .replace('__ICONS__', json.dumps(FAV_ICONS, ensure_ascii=False))) return title + head + sortbar + f'
{rows}
' + sort_js + fav_js def _ret_chart_svg(series, idx_series) -> str: """일별 누적수익률 라인차트 — 전략(실선) vs KOSPI(회색 점선, 같은 구간 등락률). series=[(yyyymmdd, ret%)], idx_series=[ret%|None] (series와 같은 길이).""" if not series: return '
아직 수익률 이력이 없어요 — 매 거래일 하나씩 쌓입니다
' w, h = 720, 230 pl, pr, pt, pb = 50, 14, 16, 28 rets = [r for _, r in series] vals = rets + [r for r in idx_series if r is not None] + [0] lo, hi = min(vals), max(vals) span = (hi - lo) or 1.0 lo -= span * 0.15; hi += span * 0.15; span = hi - lo n = len(series) def X(i): return pl + (w - pl - pr) * (i / max(n - 1, 1)) def Y(v): return pt + (h - pt - pb) * (1 - (v - lo) / span) color = '#22c55e' if rets[-1] >= 0 else '#ef4444' parts = [f''] # 0% 기준선 + y축 라벨 parts.append(f'') for v in {round(hi - span * 0.15, 1), 0.0, round(lo + span * 0.15, 1)}: parts.append(f'{v:+g}%') # KOSPI 점선 ipts = [(i, r) for i, r in enumerate(idx_series) if r is not None] if len(ipts) >= 2: d = ' '.join(f'{X(i):.1f},{Y(r):.1f}' for i, r in ipts) parts.append(f'') # 전략 실선 + 점 if n >= 2: d = ' '.join(f'{X(i):.1f},{Y(r):.1f}' for i, r in enumerate(rets)) parts.append(f'') for i, r in enumerate(rets): parts.append(f'') parts.append(f'{rets[-1]:+g}%') # x축 날짜 (처음·끝, 5개 이상이면 중간도) ticks = sorted({0, n - 1} | ({n // 2} if n >= 5 else set())) for i in ticks: dt = series[i][0] parts.append(f'{dt[4:6]}/{dt[6:8]}') parts.append('') legend = (f'
━ 이 전략' + (' · ┅ KOSPI(같은 구간)' if len(ipts) >= 2 else '') + '
') return ''.join(parts) + legend def _strategy_page(vid: str) -> str: """전략 1개 자세히보기 — 수익률 차트 + 분석. /strategy?id=vN""" from . import variants as vmod, benchmark defs = vmod.load_variants() v = next((x for x in defs if x['id'] == vid), None) back = '← 비교로' style = """""" if not v: return (f'' f'{style}' f'
{back}
전략 {_esc(vid)}을 찾을 수 없어요 (삭제됐을 수 있음)
') params = v.get('params') or {} bm = _bear_mode(params) pp, tp = vmod.variant_paths(vid) pf = Portfolio.load(pp, tp) s = pf.summary() ret = s.get('total_return_pct', 0) or 0 today = datetime.now(config.KST).strftime('%Y%m%d') bench = benchmark.compare(ret, benchmark.norm_date(pf.created_at), today) # 순위·마킹·관심종목 (비교 스냅샷) cv = [c for c in (_load(vmod.COMPARE_PATH, {}) or {}).get('variants', []) if c.get('id') != 'main'] ranked = sorted(cv, key=lambda c: c['summary'].get('total_return_pct', 0) or 0, reverse=True) rank = {c['id']: n + 1 for n, c in enumerate(ranked)}.get(vid) watching = next((c.get('watching', []) or [] for c in cv if c.get('id') == vid), []) icon = {vv: ic for ic, vv in _load_favs().items()}.get(vid, '') # 모의평가(백테스트 등록 근거) — 비교탭과 동일 매칭(swept만) btres = _load(config.STATE_DIR / 'backtest_results.json', None) or {} swept = btres.get('swept_params') or [] key = _pkey({k: params[k] for k in swept if k in params} if swept else params) bt, _rk, _seen = None, 0, set() for r in btres.get('results', []): kk = _pkey(r['params']) if kk in _seen: continue _seen.add(kk); _rk += 1 if kk == key: bt = {**r, 'rank': _rk} break # 수익률 이력 (id+파라미터해시 매칭 — id 재사용 오염 방지) + KOSPI 같은 구간 ph = hashlib.md5(json.dumps(params, sort_keys=True).encode()).hexdigest()[:8] eh = _load(config.STATE_DIR / 'equity_history.json', {}) or {} series = [(d, e.get('ret', 0)) for d in sorted(eh) for e in eh[d] if e.get('id') == vid and e.get('ph') == ph] kospi = (_load(config.STATE_DIR / 'index_history.json', {}) or {}).get('KOSPI', {}) idx_series = [] base = kospi.get(series[0][0]) if series else None for d, _ in series: c = kospi.get(d) idx_series.append(round((c / base - 1) * 100, 2) if (c and base) else None) # 종합 코멘트 (결정론 — LLM 미경유) notes = [] rk_s = f'#{rank}위/{len(ranked)}' if rank else '순위 산정 전' notes.append(f'라이브 {len(series) or 1}거래일째(백테스트 제외) · 현재 {rk_s} · 누적 {ret:+.2f}%') ak = (bench.get('KOSPI') or {}).get('alpha') if ak is not None: notes.append(f'같은 기간 KOSPI 대비 {ak:+.2f}%p — ' + ('지수를 이기고 있어요' if ak > 0 else '아직 지수에 못 미쳐요' if ak < 0 else '지수와 동률')) nt, npos = s.get('closed_trades', 0) or 0, s.get('open_positions', 0) or 0 if not nt and not npos: notes.append('아직 거래가 없어요 — 진입 조건 충족을 기다리는 중') else: if npos: notes.append(f'현재 {npos}종목 보유 중 (주식 {s.get("held_value", 0):,}원)') if nt: notes.append(f'청산 {nt}건 · 승률 {s.get("win_rate_pct")}%' + (' — 표본이 적어 아직 참고만' if nt < 5 else '')) if bt: tr_, te_ = bt['train'].get('total_return_pct', 0), bt['test'].get('total_return_pct', 0) if (tr_ - te_) > 40: notes.append('⚠ 백테스트 학습-검증 격차가 커요 — 과최적 가능성, 라이브 성적을 더 지켜봐야') notes_html = '' # 모의평가 카드 if bt: wins = ' / '.join((f"{w['ret']:+g}%" if w.get('ret') is not None else '—') for w in bt.get('windows', [])) worst = bt.get('worst_ret') bt_html = ( _kvline('전체 순위', f'#{bt["rank"]}위 (고유 조합 기준)') + _kvline('검증 수익 / 학습 수익', f'{bt["test"].get("total_return_pct"):+g}% / {bt["train"].get("total_return_pct"):+g}%') + _kvline('3분할 구간 수익', wins or '—') + _kvline('최악 구간', f'{worst:+g}%' if worst is not None else '—') + _kvline('검증 MDD / 승률 / PF', f'{bt["test"].get("mdd_pct")}% · {bt["test"].get("win_rate_pct")}% · {bt["test"].get("profit_factor")}') + '
과거 데이터 모의평가 — 다음날 시가 체결·슬리피지 반영, 절대수익이 아닌 상대 비교용
') else: bt_html = '
백테스트 기록 없음 (등록 이후 스윕 미실행)
' # 파라미터 표 prow = '' for k in sorted(params): hlp = _TUNE_HELP.get(k, {}) prow += (f'{_esc(hlp.get("t", k))}
{_esc(k)}' f'{params[k]:g}') mf_t, mf_b = _market_fit(params) watch_html = ''.join( f'
{_esc(it.get("name", ""))}' f'{_esc(it.get("reason", ""))[:34]}
' for it in watching) or '
노리는 종목 없음
' rc = '#22c55e' if ret >= 0 else '#ef4444' alpha_html = _alpha_inline(bench) return f""" {_esc(vid)} 전략 상세{style}
{back}

{rank or "—"}위 {_vid_disp(vid)} {_esc(_cat_name(params))} {_BEAR_WORD[bm]}{_relax_chip_html(params)}{_immediate_chip_html(params)}{_avg_chip_html(params)} {icon}

{_esc(_market_fit_summary(params))} · 약세장대처: {_BEAR_NAME[bm]}
{notes_html}

🧩 전략 4요소 (이름 풀이)

{_cat_detail_html(params)}

📈 수익률 추이

{_ret_chart_svg(series, idx_series)}

현재 성적

{_kvline('평가자산', f'{s.get("equity", 0):,}원')} {_kvline('수익률', f'{ret:+.2f}%')} {_kvline('예수금 / 주식', f'{s.get("cash", 0):,} / {s.get("held_value", 0):,}')} {_kvline('실현손익', f'{s.get("realized_pnl", 0):,}원')} {_kvline('거래 / 승률', f'{s.get("closed_trades", 0)}건 · {s.get("win_rate_pct")}%')} {_kvline('MDD / 보유', f'{s.get("max_drawdown_pct", 0)}% · {s.get("open_positions", 0)}종목')} {_kvline('지수대비(알파)', alpha_html)}

모의평가 (등록 근거)

{bt_html}

전략 성격 — 어떤 시장에 강한가

{_esc(mf_b)}

약세장대처 — {_BEAR_NAME[bm]}

{_esc(_BEAR_DESC[bm])}

파라미터 ({len(params)}개)

{prow}
항목

보유·관심·거래

📦 보유종목 ({s.get("open_positions", 0)}){_pos_table(pf)}
🎯 관심종목 ({len(watching)}){watch_html}
📋 거래내역 ({_trade_count(tp)}건 · {s.get("closed_trades", 0)}청산){_trades_table(_read_trades(tp))}
수익률 이력은 매 거래일 EOD 기준 1점씩 누적 · 분석은 규칙 기반(LLM 미경유)
{_CHK_HELP_JS} """ _DAY_CAL_JS = """""" def _day_trades_panel(sel_date: str | None = None) -> str: """전 전략(변이)이 거래한 종목 — 날짜별 매수/매도 집계. sel_date=YYYY-MM-DD (없으면 최근 거래일).""" from . import variants as vmod # 순위맵 (비교 스냅샷) — 전략번호 옆 현재 순위 cv = [c for c in (_load(vmod.COMPARE_PATH, {}) or {}).get('variants', []) if c.get('id') != 'main'] ranked = sorted(cv, key=lambda c: c['summary'].get('total_return_pct', 0) or 0, reverse=True) rank_map = {c['id']: n + 1 for n, c in enumerate(ranked)} # 전 전략 거래로그 1회 로드 — 날짜 목록 + 선택일 집계 동시 처리 all_rows = [] # (date, vid, record) for v in vmod.load_variants(): _, tp = vmod.variant_paths(v['id']) if not tp.exists(): continue for ln in tp.read_text().splitlines(): if not ln.strip(): continue try: r = json.loads(ln) except Exception: continue d = str(r.get('ts', ''))[:10] if d: all_rows.append((d, v['id'], r)) dates = sorted({d for d, _, _ in all_rows}, reverse=True) # 거래일만, 최신순 today = datetime.now(config.KST).strftime('%Y-%m-%d') selectable = set(dates) | {today} # 오늘은 거래 없어도 선택 가능 back = '' if sel_date not in selectable: sel_date = today # 기본 = 오늘 (거래 없으면 '이 날은 거래 없어요' 표시) # 달력형 날짜 선택 — 거래일만 밝게·클릭가능, 거래 없는 날은 어둡게. 월 이동은 JS(새로고침 X). is_today = sel_date == today cal_js = (_DAY_CAL_JS.replace('__DATES__', json.dumps(dates)) .replace('__SEL__', sel_date) .replace('__TODAY__', today)) picker = (f'
' f'📅 {sel_date}{" (오늘)" if is_today else ""} ' f'▾ 날짜 선택' f'
' f'' f'{cal_js}
') head = (f'

📋 거래목록' f'
{back}

{picker}') # 선택일 집계: (code, side) → {name, vids[], prices[], pnls[]} agg: dict = {} for d, vid, r in all_rows: if d != sel_date: continue key = (r.get('code'), r.get('side')) a = agg.setdefault(key, {'name': r.get('name', ''), 'vids': [], 'prices': [], 'pnls': [], 'reason': r.get('reason', ''), 'kinds': {}, 'time': r.get('ts', '')[11:16]}) a['vids'].append(vid) if r.get('price'): a['prices'].append(r['price']) if r.get('side') == 'SELL': if r.get('pnl_pct') is not None: a['pnls'].append(r['pnl_pct']) rsn = r.get('reason', '') or '' kind = ('익절' if '익절' in rsn else '손절' if '손절' in rsn else '추세이탈' if '이탈' in rsn else '청산') a['kinds'][kind] = a['kinds'].get(kind, 0) + 1 for a in agg.values(): a['n'] = len(a['vids']) if not agg: # 빈 날도 #day-sub·#day-rows 컨테이너 유지 — 다른 날로 fetch 전환 시 갱신되게 return (head + '
이 날은 거래가 없어요
' + '
이 날은 거래가 없어요 — 스캔이 진입 조건을 기다렸어요
') # (종목, 방향) 단위로 한 아이템 — 같은 종목에 매수·매도 동시면 2개 아이템으로 분리. # 정렬: 매도(청산) 먼저, 그다음 전략 수 많은 순. _SIDE = {'BUY': ('매수', '🟢', '#22c55e'), 'SELL': ('매도', '🔴', '#ef4444')} order = sorted(agg.items(), key=lambda kv: (0 if kv[0][1] == 'SELL' else 1, -kv[1]['n'])) n_buy = sum(1 for (_, side) in agg if side == 'BUY') n_sell = sum(1 for (_, side) in agg if side == 'SELL') sub = (f'
매수 {n_buy} · 매도 {n_sell} 아이템 (전 전략 합산, 종목·방향별 구분)
') rows = '' for (code, side), a in order: label, emo, color = _SIDE.get(side, (side, '', '#9ca3af')) ps = a['prices'] prng = (f'{min(ps):,}' if len(set(ps)) == 1 else f'{min(ps):,}~{max(ps):,}') if ps else '—' extra = '' kind_badge = '' kind_row = '' if side == 'SELL': if a['pnls']: avg = sum(a['pnls']) / len(a['pnls']) pc = '#22c55e' if avg >= 0 else '#ef4444' extra = f' · 평균 {avg:+.1f}%' # 익절/손절 분류 — 매도 사유 기반. 요약엔 대표 1개, 본문엔 전체 내역 _KC = {'익절': '#22c55e', '손절': '#ef4444', '추세이탈': '#eab308', '청산': '#9ca3af'} kinds = sorted(a['kinds'].items(), key=lambda kv: -kv[1]) if kinds: top = kinds[0][0] kind_badge = f'{top} ' parts = ' · '.join(f'{k} {c}' for k, c in kinds) kind_row = (f'
청산 유형{parts}
') vids = sorted(a['vids'], key=lambda x: rank_map.get(x, 999)) vid_str = ' '.join( f'{_vid_disp(x)}' + (f'{rank_map[x]}위' if x in rank_map else '') + '' for x in vids) rows += ( f'
' f'{_esc(a["name"])}' f'{kind_badge}' f'{emo} {label} {a["n"]}' f'
' f'
{emo} {label} {a["n"]}개 전략' f'{prng}원{extra}
' f'{kind_row}' f'
{a["time"]} · 사유' f'{_esc(a["reason"])[:30]}
' f'
{vid_str}
' f'
') return head + sub + f'
{rows}
' def _engine_status(snap, asset_line) -> tuple[str, str]: """스캔 엔진 헬스 → (점 클래스 ok/idle/bad, 토스트 상세). 장중 최근스캔=초록, 장중 멈춤=빨강, 장외=대기.""" now = datetime.now(config.KST) session = engine.is_market_session(now) ts = snap.get('scanned_at') age_min, last_str = None, '—' if ts: try: dt = datetime.fromisoformat(ts) age_min = (now - dt).total_seconds() / 60 last_str = dt.strftime('%m-%d %H:%M:%S') except Exception: pass asset = asset_line if session: if age_min is not None and age_min <= 12: return 'ok', (f"🟢 시뮬 가동 중\n장중 · 5분 주기 자동 스캔\n" f"마지막 스캔 {last_str} ({age_min:.0f}분 전)\n{asset}") agetxt = f'{age_min:.0f}분 전' if age_min is not None else '기록 없음' return 'bad', (f"🔴 스캔 멈춤 의심\n장중인데 마지막 스캔이 {agetxt}입니다.\n" f"sim-scan launchd 점검이 필요할 수 있어요.\n{asset}") return 'idle', (f"🟡 대기 중 (장 마감/장외)\n다음 개장(평일 09:00)에 자동 재개돼요.\n" f"마지막 스캔 {last_str}\n{asset}") def _pos_table(pf) -> str: """보유 포지션 펼치기 테이블 (내 계좌·비교 변이 공용).""" 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 costv = int(round(entry * 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', ' ')) rows += ( f'
' f'{_esc(p["name"])} {qty:,}주' f'{cur:,} {pnl:+.1f}%' f'
' f'
수량{qty:,}주 ({tr_n}/{tr_max}분할){scaled}
' f'
평단{entry:,.0f}원
' f'
매수금액{costv:,}원
' f'
평가금액{evalv:,}원
' f'
평가손익{pnl_amt:+,}원 ({pnl:+.1f}%)
' f'
손절{p["stop"]:,}원 ' f'({(p["stop"]/entry-1)*100:+.1f}%)
' f'
목표{p["target"]:,}원 ' f'({(p["target"]/entry-1)*100:+.1f}%){trail}
' f'
진입사유{_esc(p.get("entry_reason",""))}
' f'
진입시각{entry_at}
' f'
' ) return rows or '
보유 종목 없음
' def _trade_count(path) -> int: """거래로그(jsonl) 총 건수 (매수+매도).""" try: return sum(1 for ln in path.read_text().splitlines() if ln.strip()) except Exception: return 0 def _read_trades(path, limit=30) -> list: """거래로그(jsonl) 최근 limit건, 최신순.""" out = [] if path.exists(): for line in path.read_text().splitlines()[-limit:]: try: out.append(json.loads(line)) except Exception: pass out.reverse() return out def _trades_table(trades) -> str: """거래내역 펼치기 테이블 (내 계좌·비교 변이 공용).""" 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'
' f'
체결{r["qty"]}주 @ {r["price"]:,}원
' f'
실현손익{r.get("pnl",0):+,}원 ({r.get("pnl_pct",0):+.1f}%)
' f'
평단{r.get("entry_price",0):,}원
' f'
사유{_esc(r.get("reason",""))}
' + (f'
판단근거
{_chips(r.get("signals"))}
' if r.get('signals') else '') + '
' ) return tr_rows or '
거래 없음
' def _decision_card(d, watch_n: int = 0, held_n: int = 0, acc_detail: str = '', buy_n: int = 0) -> str: """관심종목 탭 종목 카드 (일반·제외목록 공용). watch_n=관심(대기+매수준비) 총합, buy_n=매수준비중, held_n=보유. acc_detail 이 있으면 배지 클릭 시 '어느 전략이 보는지' 목록 토스트.""" chips = _chips(d.get('checks', [])) src = '/'.join(ORIGIN_LABEL.get(s, s) for s in d.get('sources', [])) acc = '' wait_n = max(0, watch_n - buy_n) # 대기중 = 관심 총합 − 매수준비중 if watch_n or held_n: # 🟢 매수준비중 · 🟡 대기중 · 🛒 보유 — 0인 항목은 생략 parts = ([f'🟢{buy_n}'] if buy_n else []) \ + ([f'🟡{wait_n}'] if wait_n else []) \ + ([f'🛒{held_n}'] if held_n else []) attrs = (f' data-title="{_esc(d.get("name"))} — 보는 전략" data-detail="{_esc(acc_detail)}"' if acc_detail else '') acc = f' {" ".join(parts)}' 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'
' f'' f'
') return ( f'
' f'{_esc(d.get("name"))} {_esc(src)}{acc}' f'{price_h}{pnl_h}{del_btn}
' f'
{_esc(d.get("reason",""))}
' f'
{chips}
' ) def render_html(view: str | None = None, day_date: str | None = None, tab: str | None = None, sub: str | None = None, wsub: str | None = None) -> str: snap = _load(config.LAST_SCAN_PATH, {}) from . import variants as vmod _cv = [c for c in (_load(vmod.COMPARE_PATH, {}) or {}).get('variants', []) if c.get('id') != 'main'] # 구버전 스냅샷의 main 잔재 무시 (기준전략 폐지) _ranked = sorted(_cv, key=lambda c: c['summary'].get('total_return_pct', 0) or 0, reverse=True) rank_map = {c.get('id'): n + 1 for n, c in enumerate(_ranked)} fav_map = _load_favs() # {아이콘: vid} icon_of = {v: ic for ic, v in fav_map.items()} # vid → 아이콘 favs = [fav_map[ic] for ic in FAV_ICONS if fav_map.get(ic) in rank_map] # 아이콘 순 = 우선순위 # 관심종목 탭 시각 — ?view 지정 > 첫 마킹(아이콘 순) > 현재 1위. (기준전략 폐지 — 2026-06-10) view = (view or '').strip() if view in ('main', 'M') or view not in rank_map: view = favs[0] if favs else (_ranked[0]['id'] if _ranked else '') view_label = _vid_disp(view) if view else '기준판단' judge_snap = snap # 폴백: params.json 판단 스냅샷 if view: try: js = engine.judge_view(view) if js: judge_snap = js except Exception: pass scanned = snap.get('scanned_at', '')[:19].replace('T', ' ') n_acc = len(_cv) rets = [c['summary'].get('total_return_pct', 0) or 0 for c in _cv] asset_line = (f'전략 {n_acc}개 운영 · 평균 {sum(rets)/len(rets):+.2f}% · 1위 {max(rets):+.2f}%' if rets else '운영 전략 없음') hcls, hdetail = _engine_status(snap, asset_line) # ---- 판단표 (state 그룹) ---- from . import universe as umod decisions = judge_snap.get('decisions', []) # 종목별 — 이 종목을 관심(매수후보)/보유 중인 계좌(메인+변이) 수 (variants_compare 스냅샷 기준) watch_n, held_n, buy_n = {}, {}, {} watchers, holders = {}, {} # code → [(id, state, reason)] / [id] — 👁 클릭 토스트용 for c in _cv: acc_id = c.get('id') for it in c.get('watching', []) or []: cd = it.get('code') watch_n[cd] = watch_n.get(cd, 0) + 1 watchers.setdefault(cd, []).append((acc_id, it.get('state'), it.get('reason', ''))) if it.get('state') in ('BUY', 'ADD'): buy_n[cd] = buy_n.get(cd, 0) + 1 for cd in c.get('held', []) or []: held_n[cd] = held_n.get(cd, 0) + 1 holders.setdefault(cd, []).append(acc_id) def _acc_sort(vid): return int(vid[1:]) if vid.startswith('v') and vid[1:].isdigit() else 999 _view_dec = {d.get('code'): d for d in decisions} _STN = {'BUY': '🟢 매수 직전', 'ADD': '🟢 추격 대기', 'WAIT': '🟡 대기', 'HOLD': '📦 보유', 'SKIP': '⚪ 매수 안 함', 'NODATA': '⚪ 데이터 없음'} def _acc_detail(code): """이 종목을 보유/관심 중인 전략 목록 — 👁🛒 클릭 토스트 본문. 맨 위에 현재 시각 전략의 판단을 항상 표시 (목록에 없으면 왜 빠졌는지 보이게).""" lines = [] vd = _view_dec.get(code) if view and vd: rk = rank_map.get(view) rk_s = f'({rk}위)' if rk else '' st = vd.get('state') lines.append(f'▶ 시각 {_vid_disp(view)}{rk_s} — {_STN.get(st, st)}: ' f'{(vd.get("reason") or "")[:38]}') lines.append('') hs = sorted(holders.get(code) or [], key=_acc_sort) if hs: lines.append('🛒 보유 ' + str(len(hs)) + '개: ' + ', '.join(f'{_vid_disp(h)}({rank_map[h]}위)' if h in rank_map else _vid_disp(h) for h in hs)) ws = sorted(watchers.get(code) or [], key=lambda x: _acc_sort(x[0])) if ws: if lines: lines.append('') _nb = sum(1 for _, st, _ in ws if st in ('BUY', 'ADD')) _nw = len(ws) - _nb lines.append(f'관심 {len(ws)}개 — 🟢매수준비 {_nb} · 🟡대기 {_nw}:') for vid, st, rsn in ws: em = '🟢' if st in ('BUY', 'ADD') else '🟡' rk = rank_map.get(vid) rk_s = f'({rk}위)' if rk else '' pre = '▶ ' if vid == view else '' lines.append(f'{pre}{em} {_vid_disp(vid)}{rk_s} — {(rsn or "")[:38]}') return '\n'.join(lines) # 상태 그룹 = 전체 전략(메인+변이) 합산 — 한 계좌라도 보유→보유 / 매수직전→매수 / 대기→대기. # 아무 계좌도 안 보면 기준전략(M) 판정. 카드 체크리스트·사유는 기준전략 시각 유지(합산 정보는 사유 앞에 표시). # 출처(origin) 맵 — auto(자동편입) 종목은 별도 섹션으로 분리 _wl0 = umod._load_watchlist() origin_map = {c: e.get('origin', '') for c, e in _wl0.items()} by_state = {} # 내 종목(워치·관심·보유·수동) by_state_auto = {} # 자동편입(auto) — 하단 별도 excluded = [] # 매매 안 하는 종목(ETF/ETN 등) → 하단 제외목록 for d in decisions: if umod.is_etf(d.get('name', '')): excluded.append(d) continue code = d.get('code') if held_n.get(code): st = 'HOLD' elif buy_n.get(code): st = 'BUY' elif watch_n.get(code): st = 'WAIT' else: st = d.get('state', 'NODATA') if st != d.get('state'): bits = [] if held_n.get(code): bits.append(f'{held_n[code]}개 전략 보유') if buy_n.get(code): bits.append(f'{buy_n[code]}개 전략 매수 직전') elif watch_n.get(code): bits.append(f'{watch_n[code]}개 전략 매수 대기') d = {**d, 'reason': ' · '.join(bits) + f' — {view_label}: {d.get("reason", "")}'} # 출처대로만 분류 — auto(자동편입)는 변이가 페이퍼로 보유 중이어도 자동편입 탭에 (보유는 그 안 '보유' 그룹에 표시) tgt = by_state_auto if origin_map.get(code) == 'auto' else by_state tgt.setdefault(st, []).append(d) def _state_blocks(bucket): # 상태별 접이식 — 액션 상태(보유·추격·매수·매도·대기)는 펼침, 제외·데이터없음은 접힘(노이즈) _open = {'HOLD', 'ADD', 'BUY', 'SELL', 'WAIT'} out = '' for st in STATE_ORDER: items = bucket.get(st, []) if not items: continue label, color = STATE_BADGE.get(st, (st, '#888')) cards = ''.join(_decision_card(d, watch_n.get(d.get('code'), 0), held_n.get(d.get('code'), 0), _acc_detail(d.get('code')), buy_n.get(d.get('code'), 0)) for d in items) op = ' open' if st in _open else '' out += (f'
' f'{label} {len(items)}' f'
{cards}
') return out mine_n = sum(len(v) for v in by_state.values()) judge_html = _state_blocks(by_state) # 관찰목록 중 아직 스캔 안 된(방금 추가 등) 종목 → '스캔 대기'로 즉시 표시 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'
' f'' f'
') judge_html += ( f'
' f'{_esc(e.get("name") or code)} {_esc(src)}' f'다음 스캔 대기{del_btn}
' ) # 자동편입(auto) 종목 → 상단 서브탭 '자동편입'으로 분리 (별도 HTML) auto_n = sum(len(v) for v in by_state_auto.values()) judge_html_auto = (_state_blocks(by_state_auto) if auto_n else '
자동편입된 종목이 없어요 (매일 장 마감 후 키움 순위정보로 편입)
') # 매매 안 하는 종목(ETF/ETN 등) → 하단 접이식 제외목록 (내 종목 쪽에 둠) if excluded: judge_html += ( f'
' f'🚫 매매 제외 (ETF·ETN 등) {len(excluded)}' f'스캔은 하되 매수 안 함' f'
' + ''.join( _decision_card(d, watch_n.get(d.get('code'), 0), held_n.get(d.get('code'), 0), _acc_detail(d.get('code')), buy_n.get(d.get('code'), 0)) for d in excluded) + '
' ) # 관심종목 탭 시각 선택 드롭다운 (M + 변이 전체) def _rk(vid): r = rank_map.get(vid) return f'{r}위 ' if r else '' opts = '' for vv in vmod.load_variants(): _sel = ' selected' if view == vv['id'] else '' _fav = f'{icon_of[vv["id"]]} ' if vv['id'] in icon_of else '' _vp = vv.get('params') or {} _rx = ' 완화' if (not int(_vp.get('TREND_SLOPE_GATE', 1)) or not int(_vp.get('MID_ARRAY_GATE', 1))) else '' _im = ' 즉시' if int(_vp.get('IMMEDIATE_ENTRY', 0)) else '' _av = ' 물타기' if int(_vp.get('AVERAGING_ENTRY', 0)) else '' opts += (f'') view_sel = (f'
체크리스트 시각: ' f'
') mkt = snap.get('market_ok', {}) mkt_h = ' · '.join(f'{k} {"양호" if v else "약세"}' for k, v in mkt.items()) or '—' market_html = _market_panel() bt_html = _backtest_panel() cmp_html = _compare_panel() day_html = _day_trades_panel(day_date) # 활성 탭/서브탭을 서버에서 미리 checked — JS가 로드 후 바꾸며 생기던 깜빡임 제거. tab = tab if tab in ('lab', 'watch', 'market') else 'lab' sub = sub if sub in ('cmp', 'bt', 'day') else 'cmp' wsub = wsub if wsub in ('mine', 'auto') else 'mine' ck = {f'tab-{tab}': ' checked', f'sub-{sub}': ' checked', f'wsub-{wsub}': ' checked'} def _ck(rid): return ck.get(rid, '') return f""" 레이 시뮬 — 자동매매 판단

🪎 레이 자동매매 시뮬레이션

마지막 스캔 {scanned or '—'} · 시장 {mkt_h}

관찰 중 종목 ({len(decisions)})

상태 그룹은 전체 전략(메인+변이) 합산 — 한 전략이라도 보유/매수대기면 그 그룹으로 · 배지: 🟢매수준비 · 🟡대기 · 🛒보유 전략 수 (탭하면 어떤 전략인지)
종목명 일부만 입력해도 후보가 떠요 · 후보를 누르면 추가 · 자산(비하이브 워치·관심·보유)은 자동 편입 · 삭제는 카드의 ✕ (수동만) · 보유 중이면 삭제해도 청산까지 추적
{view_sel}
{judge_html}
{judge_html_auto}

시장 동향

{market_html}
{cmp_html}

백테스트 순위표

{bt_html}
{day_html}
60초 자동 새로고침 · 당겨서 새로고침 · sim/ 가상 엔진 (실주문 아님)
{PTR_HTML} {REFRESH_JS} {_WATCH_SEARCH_JS} {_CHK_HELP_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 == '/strategy': vid = (urllib.parse.parse_qs(parsed.query).get('id') or [''])[0] body = _strategy_page(vid).encode('utf-8') self.send_response(200) self.send_header('Content-Type', 'text/html; charset=utf-8') self.send_header('Cache-Control', 'no-store') 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 _qs = urllib.parse.parse_qs(parsed.query) _view = (_qs.get('view') or ['main'])[0] _dd = (_qs.get('d') or [None])[0] _tab = (_qs.get('t') or [None])[0] _sub = (_qs.get('s') or [None])[0] _wsub = (_qs.get('w') or [None])[0] # 레거시 ?t=bt/cmp/day (서브탭이 t로 옴) 하위호환 — REFRESH_JS와 동일 규칙 if _tab in ('bt', 'cmp', 'day'): _sub, _tab = _tab, 'lab' if _tab in ('acct', 'tune'): _tab = 'lab' body = render_html(view=_view, day_date=_dd, tab=_tab, sub=_sub, wsub=_wsub).encode('utf-8') self.send_response(200) self.send_header('Content-Type', 'text/html; charset=utf-8') # 캐시 금지 — Safari 휴리스틱 캐싱이 옛 HTML/JS를 재사용해 탭 복원 로직이 구버전으로 돌던 문제 방지 self.send_header('Cache-Control', 'no-store') 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 == '/fav_set': # 관심 전략 마킹 — {아이콘: vid}, 아이콘 하나당 전략 1개 (다른 전략에 쓰면 옮겨옴) 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()} vid = (form.get('id') or '').strip() icon = (form.get('icon') or '').strip() favs = {ic: v for ic, v in _load_favs().items() if v != vid} # 전략당 아이콘 1개 if vid and icon in FAV_ICONS: favs[icon] = vid # 기존 주인 교체(이동) _save_favs(favs) self.send_response(303) self.send_header('Location', '/?t=lab&s=cmp') self.end_headers() return if self.path in ('/variants_add', '/variants_reset_one', '/variants_remove', '/variants_clear_all'): 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=bt&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']) elif self.path == '/variants_clear_all': vmod.clear() # 등록 변이 전부 삭제 (메인은 variants.json에 없어 보존) 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()