feat(kevin): KevinStrategy standalone decision logic
Some checks failed
ci/woodpecker/push/woodpecker Pipeline failed

Stateless: mention + account_state -> KevinDecision. Conviction-weighted
sizing, time_horizon-derived hold periods, hard per-ticker cap. The
bridge and the backtest mini-engine both call evaluate_mention so
behaviour cannot drift.
This commit is contained in:
Viktor Barzin 2026-05-24 00:51:31 +00:00
parent c4e92b580e
commit 7dcce5ea0e
3 changed files with 517 additions and 0 deletions

View file

View file

@ -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