diff --git a/signal_v2/signal_generator.py b/signal_v2/signal_generator.py new file mode 100644 index 0000000..df1c59d --- /dev/null +++ b/signal_v2/signal_generator.py @@ -0,0 +1,215 @@ +"""Phase 4 — 매수/매도 신호 생성. + +순수 함수 generate_signals(state, dedup, settings). state 를 mutate. +""" +from __future__ import annotations +import logging +from datetime import datetime +from zoneinfo import ZoneInfo + +logger = logging.getLogger(__name__) +KST = ZoneInfo("Asia/Seoul") + +MOMENTUM_SCORES = { + "strong_up": 1.0, + "weak_up": 0.7, + "neutral": 0.5, + "weak_down": 0.3, + "strong_down": 0.0, +} + + +def generate_signals(state, dedup, settings) -> None: + """Phase 4 entry — state mutating. 매수/매도 룰 적용.""" + _evaluate_buy_signals(state, dedup, settings) + _evaluate_sell_signals(state, dedup, settings) + + +# ----- 매수 ----- + +def _evaluate_buy_signals(state, dedup, settings) -> None: + candidates = _buy_candidates(state) + for ticker, name, rank in candidates: + if not _check_buy_hard_gate(state, ticker, settings): + continue + confidence = _compute_buy_confidence(state, ticker, rank) + if confidence <= settings.confidence_threshold: + continue + if dedup.is_recent(ticker, "buy", within_hours=24): + continue + state.signals[ticker] = _build_buy_signal(state, ticker, name, rank, confidence) + dedup.record(ticker, "buy", confidence=confidence) + + +def _buy_candidates(state) -> list[tuple[str, str, int | None]]: + """screener Top-N (rank 1..N) + portfolio (rank=None).""" + candidates: list[tuple[str, str, int | None]] = [] + seen: set[str] = set() + if state.screener_preview is not None: + for i, item in enumerate(state.screener_preview.get("items", [])): + ticker = item.get("ticker") + if not ticker or ticker in seen: + continue + seen.add(ticker) + name = item.get("name", ticker) + candidates.append((ticker, name, i + 1)) + if state.portfolio is not None: + for h in state.portfolio.get("holdings", []): + ticker = h.get("ticker") + if not ticker or ticker in seen: + continue + seen.add(ticker) + candidates.append((ticker, h.get("name", ticker), None)) + return candidates + + +def _check_buy_hard_gate(state, ticker: str, settings) -> bool: + pred = state.chronos_predictions.get(ticker) + if pred is None or pred["median"] <= 0: + return False + spread = pred["q90"] - pred["q10"] + if spread >= settings.chronos_spread_threshold: + return False + momentum = state.minute_momentum.get(ticker) + if momentum != settings.min_momentum_for_buy: + return False + ap = state.asking_price.get(ticker) + if ap is None or ap["bid_ratio"] < settings.asking_bid_ratio_threshold: + return False + return True + + +def _compute_buy_confidence(state, ticker: str, rank: int | None) -> float: + pred = state.chronos_predictions[ticker] + chronos_conf = pred["conf"] + minute_score = MOMENTUM_SCORES.get(state.minute_momentum.get(ticker, "neutral"), 0.5) + screener_norm = 1 - (rank - 1) / 20 if rank is not None else 0.0 + return chronos_conf * 0.5 + minute_score * 0.3 + screener_norm * 0.2 + + +def _build_buy_signal(state, ticker: str, name: str, rank: int | None, confidence: float) -> dict: + ap = state.asking_price[ticker] + return { + "ticker": ticker, + "name": name, + "action": "buy", + "confidence_webai": confidence, + "current_price": ap["current_price"], + "avg_price": None, + "pnl_pct": None, + "context": _build_context(state, ticker, rank), + "as_of": datetime.now(KST).isoformat(), + } + + +# ----- 매도 ----- + +def _evaluate_sell_signals(state, dedup, settings) -> None: + if state.portfolio is None: + return + for holding in state.portfolio.get("holdings", []): + ticker = holding.get("ticker") + if not ticker: + continue + sell = _try_stop_loss(state, holding, settings) + if sell is None: + sell = _try_anomaly(state, holding, settings) + if sell is None: + sell = _try_take_profit(state, holding, settings) + if sell is None: + continue + if dedup.is_recent(ticker, "sell", within_hours=24): + continue + state.signals[ticker] = sell + dedup.record(ticker, "sell", confidence=sell["confidence_webai"]) + + +def _try_stop_loss(state, holding: dict, settings) -> dict | None: + pnl = holding.get("pnl_pct") + if pnl is None or pnl >= settings.stop_loss_pct: + return None + return _build_sell_signal(state, holding, confidence=1.0, reason="stop_loss") + + +def _try_take_profit(state, holding: dict, settings) -> dict | None: + pnl = holding.get("pnl_pct") + if pnl is None or pnl <= settings.take_profit_pct: + return None + return _build_sell_signal(state, holding, confidence=0.6, reason="take_profit") + + +def _try_anomaly(state, holding: dict, settings) -> dict | None: + ticker = holding["ticker"] + pred = state.chronos_predictions.get(ticker) + if pred is None or pred["median"] >= -0.01: + return None + momentum = state.minute_momentum.get(ticker) + if momentum != "strong_down": + return None + ap = state.asking_price.get(ticker) + if ap is None: + return None + if ap["bid_ratio"] > (1 - settings.asking_bid_ratio_threshold): + return None + minute_score = 1.0 - MOMENTUM_SCORES.get(momentum, 0.5) + confidence = pred["conf"] * 0.5 + minute_score * 0.3 + 1.0 * 0.2 + if confidence <= settings.confidence_threshold: + return None + return _build_sell_signal(state, holding, confidence=confidence, reason="anomaly") + + +def _build_sell_signal(state, holding: dict, confidence: float, reason: str) -> dict: + ticker = holding["ticker"] + return { + "ticker": ticker, + "name": holding.get("name", ticker), + "action": "sell", + "confidence_webai": confidence, + "current_price": holding.get("current_price"), + "avg_price": holding.get("avg_price"), + "pnl_pct": holding.get("pnl_pct"), + "context": _build_context(state, ticker, rank=None, sell_reason=reason), + "as_of": datetime.now(KST).isoformat(), + } + + +# ----- Context ----- + +def _build_context(state, ticker: str, rank: int | None, sell_reason: str | None = None) -> dict: + pred = state.chronos_predictions.get(ticker) or {} + ap = state.asking_price.get(ticker) or {} + news_item = _find_news_sentiment(state, ticker) + screener_scores = _find_screener_scores(state, ticker) + context: dict = { + "chronos_pred_1d": pred.get("median"), + "chronos_pred_conf": pred.get("conf"), + "chronos_q10": pred.get("q10"), + "chronos_q90": pred.get("q90"), + "screener_rank": rank, + "screener_scores": screener_scores, + "minute_momentum": state.minute_momentum.get(ticker), + "asking_bid_ratio": ap.get("bid_ratio"), + "news_sentiment": news_item.get("score") if news_item else None, + "news_reason": news_item.get("reason") if news_item else None, + } + if sell_reason is not None: + context["sell_reason"] = sell_reason + return context + + +def _find_news_sentiment(state, ticker: str) -> dict | None: + if state.news_sentiment is None: + return None + for item in state.news_sentiment.get("items", []): + if item.get("ticker") == ticker: + return item + return None + + +def _find_screener_scores(state, ticker: str) -> dict | None: + if state.screener_preview is None: + return None + for item in state.screener_preview.get("items", []): + if item.get("ticker") == ticker: + return item.get("scores") + return None diff --git a/signal_v2/tests/test_signal_generator.py b/signal_v2/tests/test_signal_generator.py new file mode 100644 index 0000000..7b70d03 --- /dev/null +++ b/signal_v2/tests/test_signal_generator.py @@ -0,0 +1,145 @@ +"""Tests for signal_generator.""" +from unittest.mock import MagicMock + +import pytest + +from signal_v2.signal_generator import generate_signals +from signal_v2.state import PollState + + +def _settings(**overrides): + """Build a Settings-like object for tests (avoid env).""" + defaults = dict( + stop_loss_pct=-0.07, + take_profit_pct=0.15, + chronos_spread_threshold=0.6, + asking_bid_ratio_threshold=0.6, + confidence_threshold=0.7, + min_momentum_for_buy="strong_up", + ) + defaults.update(overrides) + m = MagicMock() + for k, v in defaults.items(): + setattr(m, k, v) + return m + + +def _make_state_with_buy_candidate( + ticker="005930", name="삼성전자", + chronos_median=0.02, chronos_q10=-0.01, chronos_q90=0.04, chronos_conf=0.85, + momentum="strong_up", bid_ratio=0.7, current_price=78500, +): + state = PollState() + state.screener_preview = {"items": [{"ticker": ticker, "name": name}]} + state.chronos_predictions[ticker] = { + "median": chronos_median, "q10": chronos_q10, "q90": chronos_q90, + "conf": chronos_conf, "as_of": "2026-05-17T16:00:00+09:00", + } + state.minute_momentum[ticker] = momentum + state.asking_price[ticker] = { + "bid_total": int(bid_ratio * 1000), + "ask_total": int((1 - bid_ratio) * 1000), + "bid_ratio": bid_ratio, + "current_price": current_price, + "as_of": "2026-05-17T16:00:01+09:00", + } + return state + + +def _make_state_with_holding( + ticker="005930", name="삼성전자", + pnl_pct=0.0, avg_price=75000, current_price=75000, +): + state = PollState() + state.portfolio = {"holdings": [{ + "ticker": ticker, "name": name, + "avg_price": avg_price, "current_price": current_price, + "pnl_pct": pnl_pct, "profit_rate": pnl_pct * 100, + "quantity": 100, "broker": "키움", + }]} + state.screener_preview = {"items": []} + return state + + +@pytest.fixture +def dedup_mock(): + d = MagicMock() + d.is_recent.return_value = False + return d + + +def test_buy_signal_when_all_conditions_pass_and_confidence_high(dedup_mock): + state = _make_state_with_buy_candidate() + generate_signals(state, dedup_mock, _settings()) + assert "005930" in state.signals + sig = state.signals["005930"] + assert sig["action"] == "buy" + assert sig["confidence_webai"] > 0.7 + dedup_mock.record.assert_called() + + +def test_silent_when_chronos_median_negative(dedup_mock): + state = _make_state_with_buy_candidate(chronos_median=-0.01) + generate_signals(state, dedup_mock, _settings()) + assert "005930" not in state.signals + + +def test_silent_when_distribution_spread_too_wide(dedup_mock): + # spread = (0.5 - (-0.5)) / max(0.001, 0.001) = 1000 → > 0.6 + state = _make_state_with_buy_candidate( + chronos_median=0.001, chronos_q10=-0.5, chronos_q90=0.5, + ) + generate_signals(state, dedup_mock, _settings()) + assert "005930" not in state.signals + + +def test_silent_when_momentum_not_strong_up(dedup_mock): + state = _make_state_with_buy_candidate(momentum="weak_up") + generate_signals(state, dedup_mock, _settings()) + assert "005930" not in state.signals + + +def test_silent_when_bid_ratio_below_threshold(dedup_mock): + state = _make_state_with_buy_candidate(bid_ratio=0.5) + generate_signals(state, dedup_mock, _settings()) + assert "005930" not in state.signals + + +def test_silent_when_confidence_below_threshold(dedup_mock): + # chronos_conf low + rank=20 → confidence < 0.7 + state = _make_state_with_buy_candidate(chronos_conf=0.3) + # add 19 fake items to push 005930 rank to 20 + state.screener_preview["items"] = ( + [{"ticker": f"FAKE{i:03d}"} for i in range(19)] + + [{"ticker": "005930", "name": "삼성전자"}] + ) + generate_signals(state, dedup_mock, _settings()) + # confidence_webai = 0.3*0.5 + 1.0*0.3 + 0.05*0.2 = 0.46 < 0.7 + assert "005930" not in state.signals + + +def test_sell_signal_when_stop_loss_triggered(dedup_mock): + state = _make_state_with_holding(pnl_pct=-0.08, current_price=69000, avg_price=75000) + generate_signals(state, dedup_mock, _settings()) + assert "005930" in state.signals + sig = state.signals["005930"] + assert sig["action"] == "sell" + assert sig["confidence_webai"] == 1.0 + assert sig["pnl_pct"] == -0.08 + + +def test_sell_signal_when_take_profit_triggered(dedup_mock): + state = _make_state_with_holding(pnl_pct=0.16, current_price=87000, avg_price=75000) + generate_signals(state, dedup_mock, _settings()) + assert "005930" in state.signals + sig = state.signals["005930"] + assert sig["action"] == "sell" + assert sig["confidence_webai"] == 0.6 + + +def test_silent_when_dedup_recently_sent(dedup_mock): + state = _make_state_with_buy_candidate() + dedup_mock.is_recent.return_value = True + generate_signals(state, dedup_mock, _settings()) + assert "005930" not in state.signals + dedup_mock.record.assert_not_called()