96 lines
3.9 KiB
Python
96 lines
3.9 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)
|
|
climax_mult = ctx.get("climax_vol_mult", 3.0)
|
|
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 정합): 거래량 급증 + 반전 캔들
|
|
avg_vol20 = sma(vols, 20)
|
|
if avg_vol20 and ctx["today_volume"] >= climax_mult * avg_vol20 \
|
|
and price < ctx["day_open"]:
|
|
firing.append(_fire(ctx, "sell", "sell_climax", price, {
|
|
"vol_mult": round(ctx["today_volume"] / avg_vol20, 2),
|
|
"day_open": ctx["day_open"]})) # TODO: holdings_intel 대조
|
|
|
|
return firing
|