hyowons
244 lines
12 KiB
Python
244 lines
12 KiB
Python
"""매수/매도 판단 — 결정론적 규칙. LLM 미경유.
|
||
|
||
방향(애널·펀더) 약한 게이트 + 가점 → 타이밍(수급·기술)으로 진입/청산.
|
||
각 판단은 dashboard 용 체크리스트(checks)를 함께 반환한다.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
|
||
from . import config, indicators as ind_mod
|
||
|
||
|
||
def compute_indicators(series: list[dict]) -> dict | None:
|
||
"""series(어제까지+오늘) → 지표 묶음. 데이터 부족 시 None."""
|
||
closes = [c['close'] for c in series]
|
||
sma_long = ind_mod.sma(closes, config.SMA_LONG)
|
||
sma_short = ind_mod.sma(closes, config.SMA_SHORT)
|
||
atr = ind_mod.atr(series, config.ATR_PERIOD)
|
||
if sma_long is None or sma_short is None or atr is None:
|
||
return None
|
||
|
||
# 회전율 — 오늘 봉(라이브 시세)엔 turnover_rate 없음 → 최근 봉의 거래량/회전율로 유통주식 추정해 환산
|
||
hist = series[:-1]
|
||
turns = [c['turnover_rate'] for c in hist[-config.VOLUME_AVG_PERIOD:]
|
||
if c.get('turnover_rate')]
|
||
turnover_avg = sum(turns) / len(turns) if turns else None
|
||
est_float = None
|
||
for c in reversed(hist):
|
||
tr, v = c.get('turnover_rate'), c.get('volume')
|
||
if tr and v:
|
||
est_float = v / (tr / 100.0)
|
||
break
|
||
turnover_today = (series[-1]['volume'] / est_float * 100.0) if est_float else None
|
||
|
||
return {
|
||
'price': closes[-1],
|
||
'prev_close': closes[-2] if len(closes) >= 2 else closes[-1],
|
||
'today_volume': series[-1]['volume'],
|
||
'sma_short': sma_short,
|
||
'sma_long': sma_long,
|
||
'atr': atr,
|
||
'recent_high': ind_mod.recent_high(series, config.BREAKOUT_LOOKBACK),
|
||
'recent_low': ind_mod.recent_low(series, config.SWING_LOW_LOOKBACK),
|
||
'volume_avg': ind_mod.volume_avg(series, config.VOLUME_AVG_PERIOD),
|
||
'turnover_today': turnover_today,
|
||
'turnover_avg': turnover_avg,
|
||
}
|
||
|
||
|
||
def trend_ok(ind: dict) -> bool:
|
||
"""추세 1차 스크린 — 비싼 수급·애널 호출 전 게이트."""
|
||
return ind['price'] > ind['sma_long'] and ind['sma_short'] > ind['sma_long']
|
||
|
||
|
||
def compute_stop_target(entry: float, atr: float, recent_low: float | None) -> tuple[int, int]:
|
||
"""손절 = max(진입−ATR×2, 스윙저점) → 둘 중 위쪽(덜 손해). 목표 = 진입 + (진입−손절)×RR."""
|
||
atr_stop = entry - atr * config.STOP_ATR_MULT
|
||
stop = max(atr_stop, recent_low) if recent_low else atr_stop
|
||
stop = min(stop, entry - 1) # 손절은 진입가보다 아래여야 함
|
||
target = entry + (entry - stop) * config.RR_RATIO
|
||
return int(round(stop)), int(round(target))
|
||
|
||
|
||
def _analyst_checks(analyst: dict | None, price: float) -> tuple[bool, list[dict], int]:
|
||
"""애널 약한 게이트 + 가점. 반환: (게이트 통과, checks, 가점)."""
|
||
checks: list[dict] = []
|
||
bonus = 0
|
||
if not analyst:
|
||
checks.append({'label': '애널', 'ok': None, 'detail': '데이터 없음(통과)'})
|
||
return True, checks, 0
|
||
|
||
gate_ok = True
|
||
op = analyst.get('opinion')
|
||
if op is not None:
|
||
sell = op <= config.ANALYST_SELL_OPINION_MAX
|
||
checks.append({'label': '투자의견', 'ok': not sell,
|
||
'detail': f'{op:.1f}/5' + (' 매도권' if sell else '')})
|
||
if sell:
|
||
gate_ok = False
|
||
up = analyst.get('upside_pct')
|
||
if up is not None:
|
||
ok = up >= config.ANALYST_MIN_UPSIDE_PCT
|
||
checks.append({'label': '상승여력', 'ok': ok, 'detail': f'{up:+.0f}%'})
|
||
if not ok:
|
||
gate_ok = False
|
||
if analyst.get('revision_up'):
|
||
checks.append({'label': '목표가 리비전', 'ok': True, 'detail': '상향 ➕'})
|
||
bonus += 1
|
||
if analyst.get('surprise_pos'):
|
||
checks.append({'label': '어닝 서프라이즈', 'ok': True, 'detail': '+ ➕'})
|
||
bonus += 1
|
||
return gate_ok, checks, bonus
|
||
|
||
|
||
def evaluate_candidate(code, name, sources, ind, flow, analyst, market_ok) -> dict:
|
||
"""미보유 종목 매수 판단. flow/analyst 는 engine 이 trend 통과 후 주입."""
|
||
price = ind['price']
|
||
checks: list[dict] = []
|
||
|
||
# 1. 추세
|
||
checks.append({'label': '추세(20일선 위)', 'ok': price > ind['sma_long'],
|
||
'detail': f"{price:,} vs {ind['sma_long']:,.0f}"})
|
||
checks.append({'label': '정배열(5>20)', 'ok': ind['sma_short'] > ind['sma_long'],
|
||
'detail': f"{ind['sma_short']:,.0f} / {ind['sma_long']:,.0f}"})
|
||
|
||
# 2. 수급
|
||
fnet = flow.get('foreign', 0) if flow else 0
|
||
inet = flow.get('institution', 0) if flow else 0
|
||
supply_ok = fnet > 0 or inet > 0
|
||
checks.append({'label': f'수급({config.FLOW_DAYS}일 외/기)', 'ok': supply_ok,
|
||
'detail': f'외 {fnet:+,} · 기 {inet:+,} (천주)'})
|
||
|
||
# 3. 애널 게이트+가점
|
||
gate_ok, acks, bonus = _analyst_checks(analyst, price)
|
||
checks.extend(acks)
|
||
|
||
base = {'code': code, 'name': name, 'sources': sources, 'price': price,
|
||
'checks': checks, 'action': None, 'buy_path': None, 'buy_price': None,
|
||
'plan_stop': None, 'plan_target': None, 'bonus': bonus}
|
||
|
||
if not gate_ok:
|
||
return {**base, 'state': 'SKIP', 'reason': '애널 게이트 제외'}
|
||
if not market_ok:
|
||
return {**base, 'state': 'WAIT', 'reason': '시장 약세 — 신규매수 보류'}
|
||
if not supply_ok:
|
||
return {**base, 'state': 'WAIT', 'reason': '수급 미충족'}
|
||
|
||
# 4. 과열 보류 — 회전율이 평균 대비 과도하면 돌파·눌림목 불문 신규매수 보류
|
||
t_today, t_avg = ind.get('turnover_today'), ind.get('turnover_avg')
|
||
overheated = bool(t_today and t_avg and t_today >= t_avg * config.TURNOVER_OVERHEAT_MULT)
|
||
if t_today is not None and t_avg:
|
||
checks.append({'label': '과열 아님(회전율)', 'ok': not overheated,
|
||
'detail': f"{t_today:.2f}% vs 평균 {t_avg:.2f}% (보류 {config.TURNOVER_OVERHEAT_MULT:.0f}배↑)"})
|
||
if overheated:
|
||
return {**base, 'state': 'WAIT',
|
||
'reason': f'과열(회전율 {t_today:.1f}% ≥ 평균×{config.TURNOVER_OVERHEAT_MULT:.0f}) — 신규매수 보류'}
|
||
|
||
# 5. 매수 경로 — 돌파 우선 (전고점 + 거래량 급증 + 회전율 급증 모두 충족)
|
||
price_break = bool(ind['recent_high'] and price > ind['recent_high'])
|
||
vol_mult_ok = bool(ind['volume_avg'] and ind['today_volume'] >= ind['volume_avg'] * config.VOLUME_BREAKOUT_MULT)
|
||
turn_ok = (t_today is None or not t_avg) or (t_today >= t_avg * config.TURNOVER_BREAKOUT_MULT)
|
||
breakout = price_break and vol_mult_ok and turn_ok
|
||
|
||
checks.append({'label': '돌파(전고점)', 'ok': price_break,
|
||
'detail': f"고점 {ind['recent_high']:,.0f}" if ind['recent_high'] else '—'})
|
||
checks.append({'label': '거래량 급증', 'ok': vol_mult_ok,
|
||
'detail': f"{ind['today_volume']:,} vs 평균 {ind['volume_avg']:,.0f}" if ind['volume_avg'] else '—'})
|
||
if t_today is not None and t_avg:
|
||
checks.append({'label': '회전율 급증', 'ok': bool(turn_ok),
|
||
'detail': f"{t_today:.2f}% vs 평균 {t_avg:.2f}%"})
|
||
else:
|
||
checks.append({'label': '회전율', 'ok': None, 'detail': '데이터 없음(통과)'})
|
||
|
||
if breakout:
|
||
return {**base, 'state': 'BUY', 'action': 'buy', 'buy_path': 'breakout',
|
||
'buy_price': price, 'reason': '거래량·회전율 동반 전고점 돌파'}
|
||
|
||
# 6. 눌림목 지정가
|
||
limit = max(ind['sma_long'], ind['prev_close'] - ind['atr'] * config.PULLBACK_ATR_MULT)
|
||
limit = int(round(limit))
|
||
pulled = price <= limit
|
||
checks.append({'label': '눌림목 도달', 'ok': pulled,
|
||
'detail': f'지정가 {limit:,} (현재 {price:,})'})
|
||
if pulled:
|
||
# 현재가가 이미 지정가 이하 → 시장에서 현재가(체결가)로 매수
|
||
return {**base, 'state': 'BUY', 'action': 'buy', 'buy_path': 'pullback',
|
||
'buy_price': price, 'watch_price': limit,
|
||
'reason': f'눌림목 지정가 {limit:,} 도달'}
|
||
|
||
return {**base, 'state': 'WAIT', 'watch_price': limit,
|
||
'reason': f'눌림목 대기 — 지정가 {limit:,}'}
|
||
|
||
|
||
def evaluate_holding(position, ind, flow, analyst) -> dict:
|
||
"""보유 종목 판단. 우선순위: 손절/추세이탈(전량) → 분할익절(scale_out) → 추격매수(add) → 유지.
|
||
|
||
position 의 stop/peak/trailing/scaled_out 갱신값도 position_update 로 함께 반환.
|
||
"""
|
||
price = ind['price']
|
||
entry = position['entry_price']
|
||
stop = position['stop']
|
||
target = position['target']
|
||
peak = max(position.get('peak', entry), price)
|
||
trailing_on = position.get('trailing_on', False)
|
||
scaled_out = position.get('scaled_out', False)
|
||
tranches = position.get('tranches', 1)
|
||
last_add = position.get('last_add_price', entry)
|
||
|
||
if trailing_on:
|
||
atarget = (analyst or {}).get('target_price')
|
||
mult = config.TRAIL_ATR_MULT_TIGHT if (atarget and price >= atarget) else config.TRAIL_ATR_MULT
|
||
stop = max(stop, int(round(peak - ind['atr'] * mult)))
|
||
|
||
fnet = flow.get('foreign', 0) if flow else 0
|
||
inet = flow.get('institution', 0) if flow else 0
|
||
supply_ok = fnet > 0 or inet > 0
|
||
trend_up = price > ind['sma_long'] and ind['sma_short'] > ind['sma_long']
|
||
|
||
add_trigger = int(round(last_add + ind['atr'] * config.ADD_ATR_MULT))
|
||
can_add = tranches < config.ENTRY_TRANCHES and not trailing_on and not scaled_out
|
||
|
||
checks = [
|
||
{'label': '손절선', 'ok': price > stop, 'detail': f'{stop:,} (현재 {price:,})'},
|
||
{'label': '목표/트레일링', 'ok': None,
|
||
'detail': ('트레일링 ON' if trailing_on else
|
||
('일부익절 완료' if scaled_out else f'목표 {target:,}'))},
|
||
{'label': '추세(20일선)', 'ok': price >= ind['sma_long'],
|
||
'detail': f"{ind['sma_long']:,.0f}"},
|
||
{'label': '수급(외/기)', 'ok': supply_ok, 'detail': f'외 {fnet:+,} · 기 {inet:+,}'},
|
||
{'label': f'분할({tranches}/{config.ENTRY_TRANCHES})', 'ok': None,
|
||
'detail': (f'추가 트리거 {add_trigger:,}' if can_add else '추가 종료')},
|
||
]
|
||
|
||
upd = {'stop': stop, 'peak': peak, 'trailing_on': trailing_on, 'scaled_out': scaled_out}
|
||
pnl_pct = (price / entry - 1) * 100
|
||
|
||
base = {'code': position['code'], 'name': position['name'], 'sources': position.get('sources', []),
|
||
'price': price, 'checks': checks, 'action': None, 'state': 'HOLD',
|
||
'position_update': upd, 'pnl_pct': pnl_pct}
|
||
|
||
# 1. 손절·추세이탈 — 전량
|
||
if price <= stop:
|
||
reason = '트레일링 익절' if (trailing_on and price > entry) else '손절'
|
||
return {**base, 'state': 'SELL', 'action': 'sell', 'sell_frac': 1.0, 'reason': reason}
|
||
if price < ind['sma_long'] and fnet < 0 and inet < 0:
|
||
return {**base, 'state': 'SELL', 'action': 'sell', 'sell_frac': 1.0, 'reason': '추세·수급 이탈'}
|
||
|
||
# 2. 분할매도 — 목표 도달 & 미익절. 일부 익절 후 잔량은 트레일링으로 전환
|
||
if target and price >= target and not scaled_out:
|
||
upd['trailing_on'] = True
|
||
if config.SCALE_OUT_FRAC > 0:
|
||
upd['scaled_out'] = True
|
||
return {**base, 'state': 'SELL', 'action': 'scale_out',
|
||
'sell_frac': config.SCALE_OUT_FRAC,
|
||
'reason': f'목표 도달 — {config.SCALE_OUT_FRAC * 100:.0f}% 분할익절'}
|
||
return {**base, 'reason': '목표 도달 — 트레일링 전환'}
|
||
|
||
# 3. 추격매수 — 트랜치 여유 & 추세 지속 & 수급 & 직전 진입가 + ATR 돌파
|
||
if can_add and trend_up and supply_ok and price >= add_trigger:
|
||
return {**base, 'state': 'ADD', 'action': 'add',
|
||
'reason': f'추격매수 — {tranches + 1}/{config.ENTRY_TRANCHES}차 (추세 지속)'}
|
||
|
||
return {**base, 'reason': '보유 유지'}
|