gahusb
5dbb11ac83
BE 회신(holdings_intel.py:109-118)에 맞춰 반전 기준을 price<day_open → price<day_high×climax_close_pct(윗꼬리)로 변경. - kis_client.get_quote에 day_high(stck_hgpr) 추가 - monitor._build_ctx가 day_high를 ctx로 전달 - climax_vol_x·climax_close_pct를 monitor-set exit_params에서 읽기 (fallback: TM_CLIMAX_VOL_MULT/0.97) - 테스트 36/36 (climax exit_params 2건 추가) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01N83vbXEA8h83GMXQcg8fxD
100 lines
4.2 KiB
Python
100 lines
4.2 KiB
Python
"""§6 조건 로직 (순수). ctx + params → firing 리스트."""
|
||
from __future__ import annotations
|
||
|
||
from indicators import sma, rsi_series, highest_high
|
||
|
||
|
||
def _fire(ctx: dict, kind: str, condition: str, price: float, detail: dict) -> dict:
|
||
return {
|
||
"ticker": ctx["ticker"], "kind": kind,
|
||
"condition": condition, "price": price, "detail": detail,
|
||
}
|
||
|
||
|
||
def evaluate_buy(ctx: dict, params: dict) -> list[dict]:
|
||
price = ctx["price"]
|
||
closes, highs, lows, vols = ctx["closes"], ctx["highs"], ctx["lows"], ctx["volumes"]
|
||
rsi_os = params.get("rsi_oversold", 30)
|
||
vol_mult = params.get("breakout_vol_mult", 1.5)
|
||
pullback = params.get("pullback_pct", 0.02)
|
||
firing: list[dict] = []
|
||
|
||
# buy_ma20_pullback — 정배열 + ma20 근접 저가 + 반등 복귀
|
||
ma20, ma50, ma200 = sma(closes, 20), sma(closes, 50), sma(closes, 200)
|
||
if ma20 and ma50 and ma200 and ma20 > ma50 > ma200 and len(lows) >= 3:
|
||
recent_low = min(lows[-3:])
|
||
if recent_low <= ma20 * (1 + pullback) and price > ma20:
|
||
firing.append(_fire(ctx, "buy", "buy_ma20_pullback", price, {
|
||
"ma20": round(ma20, 1), "ma50": round(ma50, 1),
|
||
"ma200": round(ma200, 1), "recent_low": recent_low,
|
||
}))
|
||
|
||
# buy_breakout — 직전 20봉 고점 돌파 + 거래량 배수
|
||
prior_high20 = highest_high(highs, 20)
|
||
avg_vol20 = sma(vols, 20)
|
||
if prior_high20 and avg_vol20 and price > prior_high20 \
|
||
and ctx["today_volume"] > vol_mult * avg_vol20:
|
||
firing.append(_fire(ctx, "buy", "buy_breakout", price, {
|
||
"prior_high_20": prior_high20,
|
||
"vol_mult": round(ctx["today_volume"] / avg_vol20, 2),
|
||
"avg_vol_20": round(avg_vol20, 0),
|
||
}))
|
||
|
||
# buy_rsi_bounce — RSI 과매도 후 반등 (무상태 재계산)
|
||
rs = rsi_series(closes, 14)
|
||
if len(rs) >= 3 and min(rs[-3:]) < rsi_os and rs[-1] > rsi_os and rs[-1] > rs[-2]:
|
||
firing.append(_fire(ctx, "buy", "buy_rsi_bounce", price, {
|
||
"rsi": round(rs[-1], 1), "rsi_prev": round(rs[-2], 1),
|
||
"rsi_oversold": rsi_os,
|
||
}))
|
||
|
||
return firing
|
||
|
||
|
||
def evaluate_sell(ctx: dict, params: dict) -> list[dict]:
|
||
price = ctx["price"]
|
||
avg = ctx.get("avg_price")
|
||
hh = ctx.get("holding_high")
|
||
closes, vols = ctx["closes"], ctx["volumes"]
|
||
stop = params.get("stop_pct", 0.08)
|
||
take = params.get("take_pct", 0.25)
|
||
trail = params.get("trailing_pct", 0.10)
|
||
firing: list[dict] = []
|
||
|
||
if avg:
|
||
pnl = (price - avg) / avg
|
||
if pnl <= -stop:
|
||
firing.append(_fire(ctx, "sell", "sell_stop_loss", price, {
|
||
"avg_price": avg, "pnl_pct": round(pnl, 4), "stop_pct": stop}))
|
||
if pnl >= take:
|
||
firing.append(_fire(ctx, "sell", "sell_take_profit", price, {
|
||
"avg_price": avg, "pnl_pct": round(pnl, 4), "take_pct": take}))
|
||
|
||
if hh and price <= hh * (1 - trail):
|
||
firing.append(_fire(ctx, "sell", "sell_trailing_stop", price, {
|
||
"holding_high": hh, "trailing_pct": trail,
|
||
"drawdown_pct": round((price - hh) / hh, 4)}))
|
||
|
||
ma50, ma200 = sma(closes, 50), sma(closes, 200)
|
||
if ma50 and price < ma50:
|
||
severity = "high" if (ma200 and price < ma200) else "normal"
|
||
firing.append(_fire(ctx, "sell", "sell_ma_break", price, {
|
||
"ma50": round(ma50, 1),
|
||
"ma200": round(ma200, 1) if ma200 else None,
|
||
"severity": severity}))
|
||
|
||
# sell_climax — holdings_intel 정합(stock/app/holdings_intel.py:109-118):
|
||
# 거래량 ≥ 20일평균 × climax_vol_x AND 종가 < 당일고가 × climax_close_pct (윗꼬리)
|
||
# 실시간이므로 day_high = 당일 세션 누적 고가(최신 1분봉 고가 아님).
|
||
climax_vol_x = params.get("climax_vol_x", ctx.get("climax_vol_mult", 3.0))
|
||
climax_close_pct = params.get("climax_close_pct", 0.97)
|
||
avg_vol20 = sma(vols, 20)
|
||
day_high = ctx.get("day_high")
|
||
if avg_vol20 and day_high and ctx["today_volume"] >= climax_vol_x * avg_vol20 \
|
||
and price < day_high * climax_close_pct:
|
||
firing.append(_fire(ctx, "sell", "sell_climax", price, {
|
||
"vol_mult": round(ctx["today_volume"] / avg_vol20, 2),
|
||
"day_high": day_high, "climax_close_pct": climax_close_pct}))
|
||
|
||
return firing
|