diff --git a/shared/strategies/kevin.py b/shared/strategies/kevin.py new file mode 100644 index 0000000..484242a --- /dev/null +++ b/shared/strategies/kevin.py @@ -0,0 +1,213 @@ +"""Standalone Kevin strategy decision logic. + +NOT a BaseStrategy subclass — the BaseStrategy signature is bar/article +driven; Kevin is event-driven on YouTube mentions. Same shape called by +both the live signal bridge and the backtest mini-engine, so behaviour +cannot drift between them. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from decimal import Decimal +from typing import Any + +from shared.schemas.kevin import ( + KevinAccountState, + KevinDecision, + KevinDecisionType, +) +from shared.schemas.meet_kevin import TickerAction, TimeHorizon + + +@dataclass(frozen=True) +class KevinStrategyConfig: + """Strategy parameters (all overridable via env-vars in the bridge).""" + + min_conviction: Decimal + max_mention_age_hours: int + base_position_pct: Decimal + min_trade_usd: Decimal + max_trade_usd: Decimal + max_per_ticker_usd: Decimal + hold_days_by_horizon: dict[str, int] + avoid_closes_longs: bool + avoid_blocks_days: int + + +class KevinStrategy: + """Pure decision function: mention + account state -> KevinDecision. + + Stateless. The bridge owns side effects (blocklist writes, Redis counters). + """ + + name: str = "kevin" + + def __init__(self, config: KevinStrategyConfig) -> None: + self.config = config + + async def evaluate_mention( + self, + mention: Any, # KevinStockMention or stub + account: KevinAccountState, + *, + effective_conviction: Decimal, + current_price: Decimal, + is_tradable: bool, + ) -> KevinDecision: + symbol = mention.symbol + action = mention.action + horizon = mention.time_horizon + + # 1. Common no-trade gates + if not is_tradable: + return KevinDecision( + decision=KevinDecisionType.NO_OP, + symbol=symbol, + rationale="not tradable on Alpaca", + ) + + if account.paused: + return KevinDecision( + decision=KevinDecisionType.NO_OP, + symbol=symbol, + rationale="trading paused (circuit breaker / drawdown halt)", + ) + + # 2. Action-specific gates + if action in (TickerAction.HOLD, TickerAction.WATCH): + return KevinDecision( + decision=KevinDecisionType.NO_OP, + symbol=symbol, + rationale=f"action={action.value} is UI-only, never trades", + ) + + # 3. SELL — close long if held, else no-op + if action == TickerAction.SELL: + if account.is_held(symbol): + return KevinDecision( + decision=KevinDecisionType.CLOSE_LONG, + symbol=symbol, + effective_conviction=effective_conviction, + rationale="kevin SELL on held position", + ) + return KevinDecision( + decision=KevinDecisionType.NO_OP, + symbol=symbol, + rationale="SELL but not held; long-only, never shorts", + ) + + # 4. AVOID — close long if held + bridge will add blocklist (side effect) + if action == TickerAction.AVOID: + if account.is_held(symbol) and self.config.avoid_closes_longs: + return KevinDecision( + decision=KevinDecisionType.CLOSE_LONG, + symbol=symbol, + effective_conviction=effective_conviction, + rationale=( + f"kevin AVOID on held position; bridge will blocklist " + f"{self.config.avoid_blocks_days}d" + ), + ) + return KevinDecision( + decision=KevinDecisionType.NO_OP, + symbol=symbol, + rationale=( + f"kevin AVOID; bridge will add to blocklist for " + f"{self.config.avoid_blocks_days}d" + ), + ) + + # 5. BUY path — full filter stack + assert action == TickerAction.BUY + + if effective_conviction < self.config.min_conviction: + return KevinDecision( + decision=KevinDecisionType.NO_OP, + symbol=symbol, + rationale=( + f"conviction {effective_conviction} below " + f"min_conviction {self.config.min_conviction}" + ), + ) + + if horizon == TimeHorizon.INTRADAY: + return KevinDecision( + decision=KevinDecisionType.NO_OP, + symbol=symbol, + rationale="intraday horizon — 3h poll cadence can't catch", + ) + + # Mention age — uses created_at if available + if hasattr(mention, "created_at") and mention.created_at is not None: + age = datetime.now(timezone.utc) - mention.created_at + if age > timedelta(hours=self.config.max_mention_age_hours): + return KevinDecision( + decision=KevinDecisionType.NO_OP, + symbol=symbol, + rationale=( + f"mention age {age} exceeds " + f"{self.config.max_mention_age_hours}h" + ), + ) + + if account.is_blocklisted(symbol): + return KevinDecision( + decision=KevinDecisionType.NO_OP, + symbol=symbol, + rationale="symbol is on blocklist (prior AVOID)", + ) + + # 6. Compute size — conviction-weighted fixed-fractional + conviction_mult = (effective_conviction - self.config.min_conviction) / ( + Decimal("1") - self.config.min_conviction + ) + target_pct = self.config.base_position_pct * ( + Decimal("0.5") + Decimal("0.5") * conviction_mult + ) + target_dollars = account.equity_usd * target_pct + target_dollars = max( + self.config.min_trade_usd, + min(target_dollars, self.config.max_trade_usd), + ) + + # 7. Per-ticker cap absorbs multi-mention boost + already_held_usd = account.held_positions.get(symbol, Decimal("0")) + headroom = self.config.max_per_ticker_usd - already_held_usd + if headroom <= 0: + return KevinDecision( + decision=KevinDecisionType.NO_OP, + symbol=symbol, + rationale=f"at per-ticker cap (${already_held_usd} already)", + ) + target_dollars = min(target_dollars, headroom) + + if target_dollars < self.config.min_trade_usd: + return KevinDecision( + decision=KevinDecisionType.NO_OP, + symbol=symbol, + rationale=( + f"after cap headroom only ${target_dollars} — " + f"below ${self.config.min_trade_usd} floor" + ), + ) + + # Round to 2dp dollars + target_dollars = target_dollars.quantize(Decimal("0.01")) + + holding_days = self.config.hold_days_by_horizon.get( + horizon.value, self.config.hold_days_by_horizon["unspecified"] + ) + + return KevinDecision( + decision=KevinDecisionType.OPEN_LONG, + symbol=symbol, + target_dollars=target_dollars, + holding_days=holding_days, + effective_conviction=effective_conviction, + rationale=( + f"BUY conv={effective_conviction} -> " + f"{target_pct * 100}% target=${target_dollars} hold={holding_days}d" + ), + ) diff --git a/tests/shared/strategies/__init__.py b/tests/shared/strategies/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/shared/strategies/test_kevin_strategy.py b/tests/shared/strategies/test_kevin_strategy.py new file mode 100644 index 0000000..ad14271 --- /dev/null +++ b/tests/shared/strategies/test_kevin_strategy.py @@ -0,0 +1,304 @@ +"""KevinStrategy tests — one behaviour per test, full conviction x action x held grid.""" + +from datetime import datetime, timedelta, timezone +from decimal import Decimal + +import pytest + +from shared.schemas.kevin import KevinAccountState, KevinDecisionType +from shared.schemas.meet_kevin import TickerAction, TimeHorizon +from shared.strategies.kevin import KevinStrategy, KevinStrategyConfig + + +@pytest.fixture +def cfg() -> KevinStrategyConfig: + return KevinStrategyConfig( + min_conviction=Decimal("0.6"), + max_mention_age_hours=48, + base_position_pct=Decimal("0.04"), + min_trade_usd=Decimal("500"), + max_trade_usd=Decimal("5000"), + max_per_ticker_usd=Decimal("7500"), + hold_days_by_horizon={ + "days": 3, + "weeks": 10, + "months": 45, + "long_term": 90, + "unspecified": 10, + }, + avoid_closes_longs=True, + avoid_blocks_days=7, + ) + + +@pytest.fixture +def state() -> KevinAccountState: + return KevinAccountState( + equity_usd=Decimal("100000"), + cash_usd=Decimal("100000"), + held_positions={}, + blocklisted_symbols=set(), + daily_trade_count=0, + daily_alloc_usd=Decimal("0"), + paused=False, + ) + + +def _mention(symbol="NVDA", action="buy", conviction="0.7", horizon="weeks", age_hours=1): + """Lightweight stub matching KevinStockMention attribute access.""" + return type( + "M", + (), + { + "id": 1, + "symbol": symbol, + "action": TickerAction(action) if isinstance(action, str) else action, + "conviction": Decimal(conviction), + "time_horizon": TimeHorizon(horizon), + "rationale_quote": "test", + "created_at": datetime.now(timezone.utc) - timedelta(hours=age_hours), + }, + ) + + +async def test_buy_high_conviction_emits_open_long(cfg, state): + d = await KevinStrategy(cfg).evaluate_mention( + _mention(action="buy", conviction="0.8"), + state, + effective_conviction=Decimal("0.8"), + current_price=Decimal("100"), + is_tradable=True, + ) + assert d.decision == KevinDecisionType.OPEN_LONG + assert d.symbol == "NVDA" + # 4% * (0.5 + 0.5*(0.8-0.6)/0.4) = 4% * 0.75 = 3% of 100k = $3000 + assert d.target_dollars == Decimal("3000.00") + assert d.holding_days == 10 # weeks + + +async def test_buy_conviction_at_floor_emits_minimum_size(cfg, state): + d = await KevinStrategy(cfg).evaluate_mention( + _mention(action="buy", conviction="0.6"), + state, + effective_conviction=Decimal("0.6"), + current_price=Decimal("100"), + is_tradable=True, + ) + # 4% * 0.5 = 2% of 100k = $2000 + assert d.target_dollars == Decimal("2000.00") + + +async def test_buy_below_min_conviction_is_no_op(cfg, state): + d = await KevinStrategy(cfg).evaluate_mention( + _mention(action="buy", conviction="0.5"), + state, + effective_conviction=Decimal("0.5"), + current_price=Decimal("100"), + is_tradable=True, + ) + assert d.decision == KevinDecisionType.NO_OP + assert "min_conviction" in d.rationale.lower() + + +async def test_buy_mention_too_old_is_no_op(cfg, state): + d = await KevinStrategy(cfg).evaluate_mention( + _mention(age_hours=72), + state, + effective_conviction=Decimal("0.7"), + current_price=Decimal("100"), + is_tradable=True, + ) + assert d.decision == KevinDecisionType.NO_OP + assert "age" in d.rationale.lower() + + +async def test_buy_non_tradable_is_no_op(cfg, state): + d = await KevinStrategy(cfg).evaluate_mention( + _mention(symbol="OTCXX"), + state, + effective_conviction=Decimal("0.7"), + current_price=Decimal("1"), + is_tradable=False, + ) + assert d.decision == KevinDecisionType.NO_OP + assert "tradable" in d.rationale.lower() + + +async def test_buy_intraday_horizon_is_no_op(cfg, state): + d = await KevinStrategy(cfg).evaluate_mention( + _mention(horizon="intraday"), + state, + effective_conviction=Decimal("0.7"), + current_price=Decimal("100"), + is_tradable=True, + ) + assert d.decision == KevinDecisionType.NO_OP + + +async def test_buy_when_blocklisted_is_no_op(cfg, state): + state_b = state.model_copy(update={"blocklisted_symbols": {"NVDA"}}) + d = await KevinStrategy(cfg).evaluate_mention( + _mention(), + state_b, + effective_conviction=Decimal("0.7"), + current_price=Decimal("100"), + is_tradable=True, + ) + assert d.decision == KevinDecisionType.NO_OP + assert "blocklist" in d.rationale.lower() + + +async def test_buy_at_per_ticker_cap_is_no_op(cfg, state): + state_full = state.model_copy( + update={"held_positions": {"NVDA": Decimal("7500")}} + ) + d = await KevinStrategy(cfg).evaluate_mention( + _mention(), + state_full, + effective_conviction=Decimal("0.7"), + current_price=Decimal("100"), + is_tradable=True, + ) + assert d.decision == KevinDecisionType.NO_OP + assert "cap" in d.rationale.lower() + + +async def test_buy_held_below_cap_tops_up_to_cap(cfg, state): + state_p = state.model_copy( + update={"held_positions": {"NVDA": Decimal("5000")}} + ) + d = await KevinStrategy(cfg).evaluate_mention( + _mention(conviction="1.0"), + state_p, + effective_conviction=Decimal("1.0"), + current_price=Decimal("100"), + is_tradable=True, + ) + assert d.decision == KevinDecisionType.OPEN_LONG + # full target = $4000 (4% of 100k clamped at 5k cap), but per-ticker cap = $7500, + # already held $5000 -> top up min(4000, 2500) = $2500 + assert d.target_dollars == Decimal("2500.00") + + +async def test_sell_with_held_emits_close_long(cfg, state): + state_p = state.model_copy( + update={"held_positions": {"NVDA": Decimal("3000")}} + ) + d = await KevinStrategy(cfg).evaluate_mention( + _mention(action="sell"), + state_p, + effective_conviction=Decimal("0.7"), + current_price=Decimal("100"), + is_tradable=True, + ) + assert d.decision == KevinDecisionType.CLOSE_LONG + + +async def test_sell_without_held_is_no_op(cfg, state): + d = await KevinStrategy(cfg).evaluate_mention( + _mention(action="sell"), + state, + effective_conviction=Decimal("0.7"), + current_price=Decimal("100"), + is_tradable=True, + ) + assert d.decision == KevinDecisionType.NO_OP + + +async def test_avoid_with_held_emits_close_long(cfg, state): + state_p = state.model_copy( + update={"held_positions": {"NVDA": Decimal("3000")}} + ) + d = await KevinStrategy(cfg).evaluate_mention( + _mention(action="avoid"), + state_p, + effective_conviction=Decimal("0.7"), + current_price=Decimal("100"), + is_tradable=True, + ) + assert d.decision == KevinDecisionType.CLOSE_LONG + + +async def test_avoid_without_held_is_no_op_but_blocklists(cfg, state): + # The strategy itself returns NO_OP; the bridge applies the blocklist + # side-effect. We assert rationale reflects intent. + d = await KevinStrategy(cfg).evaluate_mention( + _mention(action="avoid"), + state, + effective_conviction=Decimal("0.7"), + current_price=Decimal("100"), + is_tradable=True, + ) + assert d.decision == KevinDecisionType.NO_OP + assert "blocklist" in d.rationale.lower() or "avoid" in d.rationale.lower() + + +async def test_hold_action_never_trades(cfg, state): + d = await KevinStrategy(cfg).evaluate_mention( + _mention(action="hold"), + state, + effective_conviction=Decimal("0.9"), + current_price=Decimal("100"), + is_tradable=True, + ) + assert d.decision == KevinDecisionType.NO_OP + + +async def test_watch_action_never_trades(cfg, state): + d = await KevinStrategy(cfg).evaluate_mention( + _mention(action="watch"), + state, + effective_conviction=Decimal("0.9"), + current_price=Decimal("100"), + is_tradable=True, + ) + assert d.decision == KevinDecisionType.NO_OP + + +async def test_size_clamps_to_min_trade_usd(cfg, state): + # Equity small enough that 2% < $500 floor + small_state = state.model_copy(update={"equity_usd": Decimal("10000")}) + d = await KevinStrategy(cfg).evaluate_mention( + _mention(conviction="0.6"), + small_state, + effective_conviction=Decimal("0.6"), + current_price=Decimal("100"), + is_tradable=True, + ) + # 2% of $10k = $200 -> clamped to $500 + assert d.target_dollars == Decimal("500.00") + + +async def test_size_clamps_to_max_trade_usd(cfg, state): + big_state = state.model_copy(update={"equity_usd": Decimal("1000000")}) + d = await KevinStrategy(cfg).evaluate_mention( + _mention(conviction="1.0"), + big_state, + effective_conviction=Decimal("1.0"), + current_price=Decimal("100"), + is_tradable=True, + ) + # 4% of $1M = $40k -> clamped to $5k + assert d.target_dollars == Decimal("5000.00") + + +async def test_horizon_long_term_maps_to_90_days(cfg, state): + d = await KevinStrategy(cfg).evaluate_mention( + _mention(horizon="long_term"), + state, + effective_conviction=Decimal("0.7"), + current_price=Decimal("100"), + is_tradable=True, + ) + assert d.holding_days == 90 + + +async def test_horizon_unspecified_defaults_to_weeks(cfg, state): + d = await KevinStrategy(cfg).evaluate_mention( + _mention(horizon="unspecified"), + state, + effective_conviction=Decimal("0.7"), + current_price=Decimal("100"), + is_tradable=True, + ) + assert d.holding_days == 10