hyowons
422 lines
18 KiB
Python
422 lines
18 KiB
Python
"""스캔 엔진 — 15분마다 universe 를 훑어 가상 매수/매도하고 판단 스냅샷을 남긴다.
|
||
|
||
흐름: universe → 배치 시세 → (보유=청산판단 / 미보유=추세스크린→수급·애널→매수판단)
|
||
→ 가상 체결 → portfolio 저장 → last_scan.json (대시보드용) 기록.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import sys
|
||
import time
|
||
from datetime import datetime, timedelta
|
||
|
||
from . import benchmark, config, data, signals, universe
|
||
from .portfolio import Portfolio
|
||
|
||
|
||
def _held_days(position, now) -> int | None:
|
||
"""보유 경과 일수(달력) — entry_at 기준. 시간손절 판단용. 파싱 실패 시 None."""
|
||
ts = position.get('entry_at')
|
||
if not ts:
|
||
return None
|
||
try:
|
||
return max(0, (now - datetime.fromisoformat(ts)).days)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _index_trend_ok() -> dict:
|
||
"""지수별 추세 — 지수 종가가 자기 SMA_LONG 위면 True. 데이터 부족 시 True(비차단)."""
|
||
out = {}
|
||
for mk in ('KOSPI', 'KOSDAQ'):
|
||
store = benchmark.series(mk)
|
||
if not store:
|
||
continue
|
||
closes = [v for _, v in sorted(store.items())]
|
||
if len(closes) < config.SMA_LONG + 1:
|
||
out[mk] = True
|
||
continue
|
||
sma = sum(closes[-config.SMA_LONG:]) / config.SMA_LONG
|
||
out[mk] = closes[-1] >= sma
|
||
return out
|
||
|
||
|
||
def _index_return(mk: str, days: int):
|
||
"""지수 N거래일 수익률(%) — 상대강도 비교용. 데이터 부족 시 None."""
|
||
s = sorted(benchmark.series(mk).items())
|
||
if len(s) > days and s[-(days + 1)][1]:
|
||
return (s[-1][1] / s[-(days + 1)][1] - 1) * 100
|
||
return None
|
||
|
||
|
||
def _index_daily_change() -> dict:
|
||
"""지수별 당일 등락률(%) — 전일 종가 대비. 데이터 부족 시 키 없음."""
|
||
out = {}
|
||
for mk in ('KOSPI', 'KOSDAQ'):
|
||
s = sorted(benchmark.series(mk).items())
|
||
if len(s) >= 2 and s[-2][1]:
|
||
out[mk] = (s[-1][1] / s[-2][1] - 1) * 100
|
||
return out
|
||
|
||
|
||
# 시장 국면 6단계 — (이모지, 라벨, 매수허용, 설명, 색상). market_ok·시장탭 배지 공용.
|
||
REGIME = {
|
||
'strong': ('🔥', '강세', True, '상승 종목 많고 지수도 20일선 위 — 추세 양호', '#16a34a'),
|
||
'rebound': ('📈', '반등', True, '상승 종목 많지만 지수는 아직 20일선 아래 — 약세 탈출 시도', '#22c55e'),
|
||
'mild': ('🌤', '완만 상승', True, '상승이 우위지만 강하진 않음', '#65a30d'),
|
||
'mixed': ('➖', '혼조', False, '상승·하락 비슷 — 방향 불명, 신규매수 보류', '#6b7280'),
|
||
'weak': ('📉', '약세', False, '하락 종목 우위 — 신규매수 보류', '#ef4444'),
|
||
'crash': ('❄️', '급락', False, '대부분 하락 + 지수 급락 — 신규매수 보류', '#b91c1c'),
|
||
}
|
||
|
||
|
||
def market_regime_map() -> dict:
|
||
"""시장별 국면(6단계) — 상승비율(폭) + 지수추세(방향) + 당일등락(강도) 조합.
|
||
상승비율 항상 최신(네이버, 실패 시 EOD). market_ok·시장탭 배지가 공용으로 사용."""
|
||
breadth = _live_breadth()
|
||
if breadth is None:
|
||
breadth = _eod_breadth()
|
||
chg = _index_daily_change()
|
||
trend = _index_trend_ok()
|
||
out = {}
|
||
for mk, v in breadth.items():
|
||
c = chg.get(mk)
|
||
up = trend.get(mk, True)
|
||
if v < config.MARKET_BREADTH_MIN: # <15%
|
||
out[mk] = 'crash' if (c is not None and c <= -2.0) else 'weak'
|
||
elif v >= config.MARKET_BREADTH_STRONG: # ≥60%
|
||
out[mk] = 'strong' if up else 'rebound'
|
||
elif v >= 0.40:
|
||
out[mk] = 'mild'
|
||
elif v >= 0.25:
|
||
out[mk] = 'mixed'
|
||
else:
|
||
out[mk] = 'weak'
|
||
return out
|
||
|
||
|
||
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
|
||
|
||
|
||
_LIVE_BREADTH_CACHE: dict[str, float] | None = None
|
||
_LIVE_BREADTH_AT: float = 0.0
|
||
_LIVE_BREADTH_TTL = 60.0
|
||
|
||
|
||
def _live_breadth() -> dict[str, float] | None:
|
||
"""장중 실시간 상승비율(rise/total) — 네이버 m.stock 1콜/시장. 60s 캐시, 실패 시 None."""
|
||
global _LIVE_BREADTH_CACHE, _LIVE_BREADTH_AT
|
||
now = time.time()
|
||
if _LIVE_BREADTH_CACHE is not None and (now - _LIVE_BREADTH_AT) < _LIVE_BREADTH_TTL:
|
||
return _LIVE_BREADTH_CACHE
|
||
try:
|
||
sys.path.insert(0, str(config.SCRIPTS))
|
||
import market_indicators_sync as mis
|
||
out: dict[str, float] = {}
|
||
for sym, lab in (('KOSPI', '코스피'), ('KOSDAQ', '코스닥')):
|
||
r = mis.fetch_market(sym, lab)
|
||
total = (r.get('rise', 0) + r.get('fall', 0) + r.get('steady', 0)) if r else 0
|
||
if not total:
|
||
return None
|
||
out[sym] = r['rise'] / total
|
||
_LIVE_BREADTH_CACHE, _LIVE_BREADTH_AT = out, now
|
||
return out
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _eod_breadth() -> dict[str, float]:
|
||
"""장외 폴백 — 마지막 EOD 누적 상승비율."""
|
||
latest = _load_jsonl_last_per_market(config.MARKET_HISTORY)
|
||
out: dict[str, float] = {}
|
||
for label, rec in latest.items():
|
||
total = rec.get('rise', 0) + rec.get('fall', 0) + rec.get('steady', 0)
|
||
breadth = rec.get('rise', 0) / 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
|
||
return out
|
||
|
||
|
||
def market_ok_map() -> dict[str, bool]:
|
||
"""시장별 신규매수 허용 여부 — 6단계 국면(market_regime_map) 기반.
|
||
강세·반등·완만상승 = 매수 허용 / 혼조·약세·급락 = 보류."""
|
||
return {mk: REGIME[g][2] for mk, g in market_regime_map().items()}
|
||
|
||
|
||
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 등 파라미터 의존 → 변이마다 재계산
|
||
idx_ret = {mk: _index_return(mk, config.REL_STRENGTH_DAYS) for mk in ('KOSPI', 'KOSDAQ')}
|
||
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,
|
||
held_days=_held_days(pf.positions[code], now))
|
||
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 universe.is_etf(name):
|
||
decisions.append({
|
||
'code': code, 'name': name, 'sources': sources, 'price': cur_price,
|
||
'state': 'SKIP', 'reason': 'ETF — 자동매매 제외(방향성 애널 부재)', 'action': None,
|
||
'checks': [],
|
||
})
|
||
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)
|
||
# 상대강도(라이브 전용): 종목 N일 수익률 − 시장지수 N일 수익률
|
||
n = config.REL_STRENGTH_DAYS
|
||
closes_c = [c['close'] for c in series]
|
||
if len(closes_c) > n and closes_c[-(n + 1)] and idx_ret.get(market) is not None:
|
||
stock_ret = (closes_c[-1] / closes_c[-(n + 1)] - 1) * 100
|
||
ind['rel_strength'] = round(stock_ret - idx_ret[market], 2)
|
||
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','')}")
|