You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
239 lines
7.8 KiB
Python
239 lines
7.8 KiB
Python
"""Unit tests for libs/backtest/allocator.py."""
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import math
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import pytest
|
|
|
|
from libs.backtest.domain import (
|
|
BacktestConfig,
|
|
Candidate,
|
|
DailyPortfolioState,
|
|
ExecutionConfig,
|
|
OpenPosition,
|
|
PlannedOrder,
|
|
PositionStatus,
|
|
RiskConfig,
|
|
SignalConfig,
|
|
UniverseConfig,
|
|
)
|
|
|
|
_UTC = ZoneInfo("UTC")
|
|
_NOW = dt.datetime(2026, 1, 5, 21, 0, tzinfo=_UTC)
|
|
_TODAY = dt.date(2026, 1, 5)
|
|
_TOMORROW = dt.date(2026, 1, 6)
|
|
|
|
|
|
def _make_candidate(**kwargs) -> Candidate:
|
|
defaults = dict(
|
|
event_id="EVT::TEST",
|
|
symbol="AAPL",
|
|
issuer_id=None,
|
|
score=0.8,
|
|
sector="Technology",
|
|
event_type="earnings",
|
|
event_timestamp=_NOW,
|
|
filing_time_bucket="post_market",
|
|
reaction_date=_TODAY,
|
|
execution_date=_TOMORROW,
|
|
entry_price_est=100.0,
|
|
avg_dollar_volume=5_000_000.0,
|
|
atr_14=2.0,
|
|
score_bucket="high",
|
|
)
|
|
defaults.update(kwargs)
|
|
return Candidate(**defaults)
|
|
|
|
|
|
def _make_portfolio_state(**kwargs) -> DailyPortfolioState:
|
|
defaults = dict(
|
|
date=_TOMORROW,
|
|
equity=100_000.0,
|
|
cash_available=100_000.0,
|
|
gross_exposure=0.0,
|
|
net_exposure=0.0,
|
|
reserved_risk_budget=0.0,
|
|
unrealized_pnl=0.0,
|
|
realized_pnl=0.0,
|
|
open_positions=[],
|
|
daily_new_risk_used=0.0,
|
|
peak_equity=100_000.0,
|
|
current_drawdown_pct=0.0,
|
|
)
|
|
defaults.update(kwargs)
|
|
return DailyPortfolioState(**defaults)
|
|
|
|
|
|
def _make_config(**kwargs) -> BacktestConfig:
|
|
defaults = dict(strategy_name="test", dataset_snapshot_id="snap_001")
|
|
defaults.update(kwargs)
|
|
return BacktestConfig(**defaults)
|
|
|
|
|
|
class TestComputeStopPrice:
|
|
def test_atr_stop(self):
|
|
from libs.backtest.allocator import compute_stop_price
|
|
|
|
c = _make_candidate(entry_price_est=100.0, atr_14=2.0)
|
|
stop = compute_stop_price(c, RiskConfig(
|
|
per_trade_risk_pct=0.01, max_daily_new_risk_pct=0.03,
|
|
max_positions=10, max_positions_per_sector=3
|
|
))
|
|
# 1.5 * ATR below price
|
|
assert stop == pytest.approx(100.0 - 1.5 * 2.0)
|
|
|
|
def test_fallback_stop_when_no_atr(self):
|
|
from libs.backtest.allocator import compute_stop_price
|
|
|
|
c = _make_candidate(entry_price_est=100.0, atr_14=None)
|
|
stop = compute_stop_price(c, RiskConfig(
|
|
per_trade_risk_pct=0.01, max_daily_new_risk_pct=0.03,
|
|
max_positions=10, max_positions_per_sector=3
|
|
))
|
|
# 2% fallback
|
|
assert stop == pytest.approx(98.0)
|
|
|
|
def test_stop_never_negative(self):
|
|
from libs.backtest.allocator import compute_stop_price
|
|
|
|
c = _make_candidate(entry_price_est=1.0, atr_14=5.0)
|
|
stop = compute_stop_price(c, RiskConfig(
|
|
per_trade_risk_pct=0.01, max_daily_new_risk_pct=0.03,
|
|
max_positions=10, max_positions_per_sector=3
|
|
))
|
|
assert stop >= 0.01
|
|
|
|
|
|
class TestComputeShares:
|
|
def test_basic(self):
|
|
from libs.backtest.allocator import compute_shares
|
|
|
|
# 1% of 100k = 1000 risk, 100-95=5 stop distance → 200 shares
|
|
shares = compute_shares(
|
|
100_000.0, 100.0, 95.0,
|
|
RiskConfig(per_trade_risk_pct=0.01, max_daily_new_risk_pct=0.03,
|
|
max_positions=10, max_positions_per_sector=3)
|
|
)
|
|
assert shares == 200
|
|
|
|
def test_always_floor(self):
|
|
from libs.backtest.allocator import compute_shares
|
|
|
|
# Result should always be floor
|
|
shares = compute_shares(
|
|
100_000.0, 100.0, 96.7,
|
|
RiskConfig(per_trade_risk_pct=0.01, max_daily_new_risk_pct=0.03,
|
|
max_positions=10, max_positions_per_sector=3)
|
|
)
|
|
# raw = 1000 / 3.3 ≈ 303.03 → floor = 303
|
|
assert shares == math.floor(1000.0 / 3.3)
|
|
|
|
def test_zero_when_stop_above_entry(self):
|
|
from libs.backtest.allocator import compute_shares
|
|
|
|
shares = compute_shares(100_000.0, 95.0, 100.0, RiskConfig(
|
|
per_trade_risk_pct=0.01, max_daily_new_risk_pct=0.03,
|
|
max_positions=10, max_positions_per_sector=3
|
|
))
|
|
assert shares == 0
|
|
|
|
|
|
class TestRunEntryGates:
|
|
def test_pass_all_gates(self):
|
|
from libs.backtest.allocator import run_entry_gates
|
|
|
|
c = _make_candidate()
|
|
ps = _make_portfolio_state()
|
|
cfg = _make_config()
|
|
assert run_entry_gates(c, ps, [], cfg) is None
|
|
|
|
def test_gate1_kill_switch(self):
|
|
from libs.backtest.allocator import run_entry_gates
|
|
|
|
c = _make_candidate()
|
|
ps = _make_portfolio_state(current_drawdown_pct=30.0)
|
|
assert run_entry_gates(c, ps, [], _make_config()) == "kill_switch_drawdown"
|
|
|
|
def test_gate2_max_positions(self):
|
|
from libs.backtest.allocator import run_entry_gates
|
|
|
|
c = _make_candidate()
|
|
ps = _make_portfolio_state()
|
|
cfg = _make_config()
|
|
cfg.risk.max_positions = 0 # impossible to add
|
|
|
|
# Mock 0 positions but max is 0
|
|
result = run_entry_gates(c, ps, [], cfg)
|
|
assert result == "max_positions_reached"
|
|
|
|
def test_gate4_sector_limit(self):
|
|
from libs.backtest.allocator import run_entry_gates
|
|
|
|
# Create a position in the same sector
|
|
plan = PlannedOrder(
|
|
candidate=_make_candidate(symbol="MSFT", sector="Technology"),
|
|
shares=10, entry_price_limit=100.0, stop_price=95.0,
|
|
target_price=110.0, risk_dollars=50.0,
|
|
)
|
|
existing_pos = OpenPosition(
|
|
position_id="p1", plan=plan, entry_date=_TODAY,
|
|
entry_price=100.0, entry_fill_slippage_bps=10.0,
|
|
current_stop=95.0, target_price=110.0, peak_price=100.0,
|
|
shares_open=10, shares_total=10,
|
|
)
|
|
cfg = _make_config()
|
|
cfg.risk.max_positions_per_sector = 1 # only 1 per sector
|
|
|
|
c = _make_candidate(symbol="AAPL", sector="Technology")
|
|
result = run_entry_gates(c, _make_portfolio_state(), [existing_pos], cfg)
|
|
assert result == "sector_limit"
|
|
|
|
def test_gate3_duplicate_symbol(self):
|
|
from libs.backtest.allocator import run_entry_gates
|
|
|
|
plan = PlannedOrder(
|
|
candidate=_make_candidate(symbol="AAPL"),
|
|
shares=10, entry_price_limit=100.0, stop_price=95.0,
|
|
target_price=110.0, risk_dollars=50.0,
|
|
)
|
|
existing = OpenPosition(
|
|
position_id="p1", plan=plan, entry_date=_TODAY,
|
|
entry_price=100.0, entry_fill_slippage_bps=10.0,
|
|
current_stop=95.0, target_price=110.0, peak_price=100.0,
|
|
shares_open=10, shares_total=10,
|
|
)
|
|
c = _make_candidate(symbol="AAPL")
|
|
result = run_entry_gates(c, _make_portfolio_state(), [existing], _make_config())
|
|
assert result == "duplicate_symbol"
|
|
|
|
def test_gate7_cooldown(self):
|
|
from libs.backtest.allocator import run_entry_gates
|
|
|
|
c = _make_candidate()
|
|
result = run_entry_gates(c, _make_portfolio_state(), [], _make_config(), cooldown_remaining=2)
|
|
assert result == "cooldown"
|
|
|
|
|
|
class TestBuildPlannedOrder:
|
|
def test_valid_order(self):
|
|
from libs.backtest.allocator import build_planned_order
|
|
|
|
c = _make_candidate(entry_price_est=100.0, atr_14=2.0)
|
|
ps = _make_portfolio_state()
|
|
order = build_planned_order(c, ps, [], _make_config())
|
|
assert order.skip_reason is None
|
|
assert order.shares > 0
|
|
assert order.stop_price < 100.0
|
|
assert order.target_price > 100.0
|
|
|
|
def test_rejected_order_has_skip_reason(self):
|
|
from libs.backtest.allocator import build_planned_order
|
|
|
|
c = _make_candidate()
|
|
ps = _make_portfolio_state(current_drawdown_pct=30.0)
|
|
order = build_planned_order(c, ps, [], _make_config())
|
|
assert order.skip_reason == "kill_switch_drawdown"
|
|
assert order.shares == 0
|