"""스캔 엔진 — 15분마다 universe 를 훑어 가상 매수/매도하고 판단 스냅샷을 남긴다. 흐름: universe → 배치 시세 → (보유=청산판단 / 미보유=추세스크린→수급·애널→매수판단) → 가상 체결 → portfolio 저장 → last_scan.json (대시보드용) 기록. """ from __future__ import annotations import json import sys from datetime import datetime, timedelta from . import config, data, signals, universe from .portfolio import Portfolio def _load_jsonl_last_per_market(path) -> dict[str, dict]: out: dict[str, dict] = {} if not path.exists(): return out try: for line in path.read_text().splitlines(): if not line.strip(): continue rec = json.loads(line) out[rec.get('market', '')] = rec except Exception: pass return out def market_ok_map() -> dict[str, bool]: """시장별 신규매수 허용 여부 — 당일 상승비율 기반 소프트 필터 (EOD 데이터라 보수적).""" latest = _load_jsonl_last_per_market(config.MARKET_HISTORY) out: dict[str, bool] = {} for label, rec in latest.items(): rise = rec.get('rise', 0) fall = rec.get('fall', 0) steady = rec.get('steady', 0) total = rise + fall + steady breadth = rise / total if total else 1.0 key = 'KOSPI' if '코스피' in (rec.get('market_label') or '') or label == 'KOSPI' else \ 'KOSDAQ' if '코스닥' in (rec.get('market_label') or '') or label == 'KOSDAQ' else label out[key] = breadth >= config.MARKET_BREADTH_MIN return out def _is_holiday(d: datetime) -> bool: try: hol = json.loads(config.HOLIDAYS_PATH.read_text()) days = hol if isinstance(hol, list) else hol.get('holidays', []) return d.strftime('%Y-%m-%d') in set(days) or d.strftime('%Y%m%d') in set(days) except Exception: return False def is_market_session(now: datetime | None = None) -> bool: now = now or datetime.now(config.KST) if now.weekday() >= 5 or _is_holiday(now): return False o = now.replace(hour=config.MARKET_OPEN[0], minute=config.MARKET_OPEN[1], second=0, microsecond=0) c = now.replace(hour=config.MARKET_CLOSE[0], minute=config.MARKET_CLOSE[1], second=0, microsecond=0) return o <= now <= c def _fill_buy_price(code: str, suggested: int) -> int: book = data.quote_book(code) if book and book.get('asks'): ask1 = book['asks'][0].get('price') if ask1: return int(ask1) return int(suggested) def _fill_sell_price(code: str, fallback: int) -> int: book = data.quote_book(code) if book and book.get('bids'): bid1 = book['bids'][0].get('price') if bid1: return int(bid1) return int(fallback) def _gather(extra_codes=None): """파라미터 무관한 라이브 데이터 1회 수집 (메인·변이 공유). 종목·시세·시장매핑. extra_codes: universe에 없지만 청산 판단을 위해 포함해야 할 보유 코드(수동 삭제된 보유분). """ uni = universe.build_universe() codes = [e['code'] for e in uni] seen = set(codes) for c in (extra_codes or set()): if c and c not in seen: uni = uni + [{'code': c, 'name': '', 'sources': ['held']}] codes.append(c) seen.add(c) quotes = data.batch_quotes(codes) cmkt = universe.code_market_map() return uni, quotes, cmkt def _run_scan(pf, uni, quotes, cmkt, now) -> dict: """주어진 포트폴리오에 대해 현재 config(파라미터) 기준 스캔·체결. 스냅샷 반환(파일 기록 X).""" mkt_ok = market_ok_map() # MARKET_BREADTH_MIN 등 파라미터 의존 → 변이마다 재계산 decisions: list[dict] = [] buys: list[dict] = [] sells: list[dict] = [] for e in uni: code, name, sources = e['code'], e['name'], e['sources'] if not name and code in pf.positions: # 보유 보강분(universe에서 빠진)은 이름 보완 name = pf.positions[code].get('name', code) quote = quotes.get(code) hist = data.candles(code, 80) series = data.build_series(hist, quote) ind = signals.compute_indicators(series) if ind is None: decisions.append({'code': code, 'name': name, 'sources': sources, 'state': 'NODATA', 'reason': '데이터 부족', 'checks': [], 'price': (quote or {}).get('price')}) continue cur_price = ind['price'] # ---- 보유: 청산 판단 ---- if code in pf.positions: pf.mark(code, cur_price) flow = data.flow_net(data.investor_flow(code)) anl = data.analyst(code, cur_price) dec = signals.evaluate_holding(pf.positions[code], ind, flow, anl) pf.apply_position_update(code, dec['position_update']) dec.update({k: pf.positions[code].get(k) for k in ('entry_price', 'qty', 'stop', 'target', 'trailing_on', 'entry_at')}) act = dec['action'] if act in ('sell', 'scale_out'): fill = _fill_sell_price(code, cur_price) rec = pf.sell(code, fill, dec['reason'], signals=dec.get('checks'), frac=dec.get('sell_frac', 1.0)) if rec: sells.append(rec) dec['fill_price'] = fill elif act == 'add': fill = _fill_buy_price(code, cur_price) rec = pf.add_tranche(code, fill, dec['reason'], signals=dec.get('checks')) if rec: # 새 평단 기준 손절·목표 재산정 (손절은 위로만 래칫) new_entry = pf.positions[code]['entry_price'] nstop, ntarget = signals.compute_stop_target(new_entry, ind['atr'], ind['recent_low']) pf.positions[code]['stop'] = max(pf.positions[code]['stop'], nstop) pf.positions[code]['target'] = ntarget buys.append(rec) dec.update({'fill_price': fill, 'entry_price': new_entry, 'qty': pf.positions[code]['qty'], 'stop': pf.positions[code]['stop'], 'target': ntarget}) decisions.append(dec) continue # ---- 미보유: 추세 스크린 → 수급·애널 → 매수 판단 ---- if not signals.trend_ok(ind): decisions.append({ 'code': code, 'name': name, 'sources': sources, 'price': cur_price, 'state': 'SKIP', 'reason': '추세 미충족', 'action': None, 'checks': [ {'label': '추세(20일선 위)', 'ok': cur_price > ind['sma_long'], 'detail': f"{cur_price:,} vs {ind['sma_long']:,.0f}"}, {'label': '정배열(5>20)', 'ok': ind['sma_short'] > ind['sma_long'], 'detail': f"{ind['sma_short']:,.0f} / {ind['sma_long']:,.0f}"}, ], }) continue flow = data.flow_net(data.investor_flow(code)) anl = data.analyst(code, cur_price) market = cmkt.get(code, '') m_ok = mkt_ok.get(market, True) dec = signals.evaluate_candidate(code, name, sources, ind, flow, anl, m_ok) if dec['action'] == 'buy' and pf.can_open(): fill = _fill_buy_price(code, dec['buy_price']) stop, target = signals.compute_stop_target(fill, ind['atr'], ind['recent_low']) rec = pf.buy(code, name, fill, sources, stop, target, dec['buy_path'], dec['reason'], signals=dec.get('checks')) if rec: buys.append(rec) dec.update({'fill_price': fill, 'plan_stop': stop, 'plan_target': target}) else: dec['state'] = 'WAIT' dec['reason'] = '매수 신호 — 현금/한도 부족' decisions.append(dec) pf.update_equity_metrics() pf.save() order = {'SELL': 0, 'ADD': 1, 'BUY': 2, 'HOLD': 3, 'WAIT': 4, 'SKIP': 5, 'NODATA': 6} decisions.sort(key=lambda d: (order.get(d.get('state'), 9), d.get('name', ''))) return { 'scanned_at': now.isoformat(), 'next_scan_hint': (now + timedelta(minutes=15)).isoformat(), 'session': is_market_session(now), 'market_ok': mkt_ok, 'summary': pf.summary(), 'counts': {'buys': len(buys), 'sells': len(sells), 'universe': len(uni)}, 'decisions': decisions, } def _attach_benchmark(snap: dict, pf, now, refresh: bool = True): """가상계좌 시작일 대비 기준지수(KOSPI/KOSDAQ) 수익·알파를 스냅샷에 부착.""" from . import benchmark if refresh: benchmark.update_today() start = benchmark.norm_date(pf.created_at) today = now.strftime('%Y%m%d') snap['benchmark'] = { 'since': start, 'to': today, 'indices': benchmark.compare(snap['summary'].get('total_return_pct'), start, today), } def scan(force: bool = False) -> dict: """메인 sim 1회 스캔 — last_scan.json 기록.""" now = datetime.now(config.KST) if not force and not is_market_session(now): return {'skipped': True, 'reason': '장외/휴장', 'at': now.isoformat()} pf = Portfolio.load() uni, quotes, cmkt = _gather(set(pf.positions)) # 보유분은 universe에서 빠져도 청산 위해 포함 snap = _run_scan(pf, uni, quotes, cmkt, now) _attach_benchmark(snap, pf, now) config.STATE_DIR.mkdir(parents=True, exist_ok=True) config.LAST_SCAN_PATH.write_text(json.dumps(snap, ensure_ascii=False, indent=2)) return snap def scan_all(force: bool = False) -> dict: """메인 + 병렬 페이퍼 변이 전부 스캔 (라이브 데이터 1회 공유). 비교 스냅샷 기록.""" from . import backtest, variants as variants_mod now = datetime.now(config.KST) if not force and not is_market_session(now): return {'skipped': True, 'reason': '장외/휴장', 'at': now.isoformat()} # 포트폴리오 먼저 로드해 보유 코드 합집합 수집 (universe에서 빠진 보유분도 청산 위해 포함) main_params = config.load_params() pf = Portfolio.load() vdefs = variants_mod.load_variants() vpfs = {v['id']: Portfolio.load(*variants_mod.variant_paths(v['id'])) for v in vdefs} held_union = set(pf.positions) for vpf in vpfs.values(): held_union |= set(vpf.positions) uni, quotes, cmkt = _gather(held_union) # 메인 (현재 params.json 튜닝) backtest.apply_params(main_params) snap = _run_scan(pf, uni, quotes, cmkt, now) _attach_benchmark(snap, pf, now) # 지수 캐시 갱신 1회 (변이는 캐시 재사용) config.STATE_DIR.mkdir(parents=True, exist_ok=True) config.LAST_SCAN_PATH.write_text(json.dumps(snap, ensure_ascii=False, indent=2)) from . import benchmark today = now.strftime('%Y%m%d') def _alpha(p): s = p.summary() return benchmark.compare(s.get('total_return_pct'), benchmark.norm_date(p.created_at), today) compare = [{'id': 'main', 'name': '메인 (현재 튜닝)', 'params': main_params, 'summary': pf.summary(), 'benchmark': _alpha(pf)}] for v in vdefs: backtest.apply_params(v.get('params') or {}) vpf = vpfs[v['id']] _run_scan(vpf, uni, quotes, cmkt, now) compare.append({'id': v['id'], 'name': v.get('name', v['id']), 'params': v.get('params') or {}, 'summary': vpf.summary(), 'benchmark': _alpha(vpf)}) compare.sort(key=lambda c: c['summary'].get('total_return_pct', 0) or 0, reverse=True) variants_mod.COMPARE_PATH.write_text(json.dumps({ 'scanned_at': now.isoformat(), 'session': is_market_session(now), 'variants': compare, }, ensure_ascii=False, indent=2)) return snap if __name__ == '__main__': force = '--force' in sys.argv snap = scan_all(force=force) if snap.get('skipped'): print(f"[skip] {snap['reason']} @ {snap['at']}") else: s = snap['summary'] print(f"자산 {s['equity']:,}원 ({s['total_return_pct']:+.2f}%) · " f"현금 {s['cash']:,} · 보유 {s['open_positions']} · " f"매수 {snap['counts']['buys']} 매도 {snap['counts']['sells']}") for d in snap['decisions']: if d['state'] in ('BUY', 'ADD', 'SELL', 'HOLD'): print(f" [{d['state']}] {d['name']:12s} {d.get('reason','')}")