diff --git a/services/trade-monitor/conditions.py b/services/trade-monitor/conditions.py new file mode 100644 index 0000000..71e8932 --- /dev/null +++ b/services/trade-monitor/conditions.py @@ -0,0 +1,95 @@ +"""§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 diff --git a/services/trade-monitor/tests/test_conditions_buy.py b/services/trade-monitor/tests/test_conditions_buy.py new file mode 100644 index 0000000..a9eb833 --- /dev/null +++ b/services/trade-monitor/tests/test_conditions_buy.py @@ -0,0 +1,66 @@ +"""evaluate_buy — 3개 매수 조건 경계.""" +from conditions import evaluate_buy + +BUY_PARAMS = {"rsi_oversold": 30, "breakout_vol_mult": 1.5, "pullback_pct": 0.02} + + +def _ctx(**over): + base = dict( + ticker="005930", name="삼성전자", price=100.0, day_open=99.0, + today_volume=1000.0, closes=[], highs=[], lows=[], volumes=[], + avg_price=None, qty=None, holding_high=None, climax_vol_mult=3.0, + ) + base.update(over) + return base + + +def _conditions(firing): + return {f["condition"] for f in firing} + + +def test_ma20_pullback_fires(): + # 정배열(ma20>ma50>ma200), 최근 저가가 ma20 근처, price가 ma20 위로 반등 + closes = [90.0] * 200 + [100.0] * 20 # ma20=100, ma50/ma200 낮음 → 정배열 + lows = [90.0] * 217 + [100.5, 100.4, 100.3] # 최근 3봉 저가 ~ma20*(1.02)=102 이하 + ctx = _ctx(price=101.0, closes=closes, highs=closes, lows=lows, + volumes=[1.0] * len(closes)) + assert "buy_ma20_pullback" in _conditions(evaluate_buy(ctx, BUY_PARAMS)) + + +def test_ma20_pullback_skips_when_not_aligned(): + closes = [100.0] * 200 + [90.0] * 20 # 역배열 + ctx = _ctx(price=91.0, closes=closes, highs=closes, lows=closes, + volumes=[1.0] * len(closes)) + assert "buy_ma20_pullback" not in _conditions(evaluate_buy(ctx, BUY_PARAMS)) + + +def test_breakout_fires(): + closes = [50.0] * 25 + highs = [60.0] * 25 # 직전 20봉 최고 60 + vols = [100.0] * 25 # avg20=100 + ctx = _ctx(price=61.0, today_volume=200.0, closes=closes, highs=highs, + lows=closes, volumes=vols) # 61>60, 200>1.5*100 + assert "buy_breakout" in _conditions(evaluate_buy(ctx, BUY_PARAMS)) + + +def test_breakout_skips_on_low_volume(): + highs = [60.0] * 25 + ctx = _ctx(price=61.0, today_volume=120.0, closes=[50.0] * 25, highs=highs, + lows=[50.0] * 25, volumes=[100.0] * 25) # 120 < 1.5*100=150 + assert "buy_breakout" not in _conditions(evaluate_buy(ctx, BUY_PARAMS)) + + +def test_rsi_bounce_fires(): + # 14봉 급락으로 RSI<30 찍고 5봉 반등하여 30 위로 복귀 + closes = [100.0] + for _ in range(14): + closes.append(closes[-1] * 0.97) # 하락 → RSI 저하 + for _ in range(5): + closes.append(closes[-1] * 1.05) # 반등 → RSI 30 위로 + ctx = _ctx(price=closes[-1], closes=closes, highs=closes, lows=closes, + volumes=[1.0] * len(closes)) + assert "buy_rsi_bounce" in _conditions(evaluate_buy(ctx, BUY_PARAMS)) + + +def test_empty_series_no_fire(): + assert evaluate_buy(_ctx(), BUY_PARAMS) == [] diff --git a/services/trade-monitor/tests/test_conditions_sell.py b/services/trade-monitor/tests/test_conditions_sell.py new file mode 100644 index 0000000..5bd5681 --- /dev/null +++ b/services/trade-monitor/tests/test_conditions_sell.py @@ -0,0 +1,69 @@ +"""evaluate_sell — 5개 매도 조건 경계.""" +from conditions import evaluate_sell + +EXIT = {"stop_pct": 0.08, "take_pct": 0.25, "trailing_pct": 0.10} + + +def _ctx(**over): + base = dict( + ticker="000660", name="SK하이닉스", price=100.0, day_open=100.0, + today_volume=100.0, closes=[100.0] * 60, highs=[100.0] * 60, + lows=[100.0] * 60, volumes=[100.0] * 60, + avg_price=100.0, qty=10, holding_high=100.0, climax_vol_mult=3.0, + ) + base.update(over) + return base + + +def _c(firing): + return {f["condition"] for f in firing} + + +def test_stop_loss_fires(): + ctx = _ctx(price=90.0, avg_price=100.0) # -10% <= -8% + assert "sell_stop_loss" in _c(evaluate_sell(ctx, EXIT)) + + +def test_stop_loss_skips_above_threshold(): + ctx = _ctx(price=95.0, avg_price=100.0) # -5% > -8% + assert "sell_stop_loss" not in _c(evaluate_sell(ctx, EXIT)) + + +def test_take_profit_fires(): + ctx = _ctx(price=130.0, avg_price=100.0) # +30% >= 25% + assert "sell_take_profit" in _c(evaluate_sell(ctx, EXIT)) + + +def test_trailing_stop_fires(): + ctx = _ctx(price=89.0, holding_high=100.0) # 89 <= 100*0.9=90 + assert "sell_trailing_stop" in _c(evaluate_sell(ctx, EXIT)) + + +def test_ma_break_severity_high(): + # price가 ma50/ma200 아래 → severity high (ma200 계산 위해 200봉 필요) + closes = [200.0] * 200 + ctx = _ctx(price=100.0, closes=closes, avg_price=100.0, holding_high=100.0) + firing = evaluate_sell(ctx, EXIT) + mb = [f for f in firing if f["condition"] == "sell_ma_break"] + assert mb and mb[0]["detail"]["severity"] == "high" + + +def test_climax_fires(): + # 거래량 3배 이상 + 종가(현재가)<시가 반전 + ctx = _ctx(price=98.0, day_open=100.0, today_volume=400.0, + volumes=[100.0] * 60) # 400 >= 3*100, 98<100 + assert "sell_climax" in _c(evaluate_sell(ctx, EXIT)) + + +def test_climax_skips_when_not_reversal(): + ctx = _ctx(price=101.0, day_open=100.0, today_volume=400.0, + volumes=[100.0] * 60) # 상승 마감 → 반전 아님 + assert "sell_climax" not in _c(evaluate_sell(ctx, EXIT)) + + +def test_no_avg_no_pnl_conditions(): + # avg_price None(보유정보 없음) → stop/take 미발화 + ctx = _ctx(price=50.0, avg_price=None, holding_high=None, + closes=[100.0] * 60) + conds = _c(evaluate_sell(ctx, EXIT)) + assert "sell_stop_loss" not in conds and "sell_take_profit" not in conds