yirugi
/
fithia2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix stale chunk_checkpoint test; add _record_trade and run_post_close tests

Browse Source 
- Fix pre-existing test failure: fake_run_sim mock was missing **kwargs for
  the vix_by_day argument added to run_orb_simulation_with_state

- Add TestRecordTrade (7 tests): verifies long/short PnL sign, R-multiple,
  DB close call, and exit_reason preservation — the direction sign bug would
  silently invert short-trade PnL

- Add TestRunPostClose (4 tests): equity accumulation, stops_hit counter,
  today-only trade filter, snapshot persistence

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
main
I Luk Kim 4 months ago
parent ba491b2a0d
commit d4daf7a951
2 changed files with 671 additions and 0 deletions
Show Stats Download Patch File Download Diff File

489
tests/unit/intraday/test_orb_research.py
Unescape Escape View File

@ -0,0 +1,489 @@
from __future__ import annotations
import datetime as dt
import pytest
from libs.backtest.domain import (
SplitResult,
WalkForwardAggregate,
WalkForwardGapStats,
WalkForwardSummary,
)
from libs.intraday.domain import (
BacktestParams,
CacheParams,
DayResult,
IntradayConfig,
ORBStrategyParams,
OutputParams,
UniverseParams,
)
from libs.intraday.orb_simulator import ORBSimulationState
import apps.intraday_bt.orb_research as orb_research
from apps.intraday_bt.orb_research import build_orb_research_context, compute_orb_overfit_score, compute_orbqs
def test_compute_orb_overfit_score_is_weighted_and_bounded() -> None:
score, breakdown = compute_orb_overfit_score(
{"retention_pct": 80.0},
{"mean_sharpe": 1.2, "cv": 0.4},
{"params": [{"plateau": 0.8}, {"plateau": 0.6}]},
{"p_value": 0.04},
)
assert 0.0 <= score <= 100.0
assert set(breakdown) == {
"is_oos_retention",
"wf_stability",
"parameter_plateau",
"candidate_permutation",
}
def test_compute_orbqs_returns_breakdown_and_activity_penalty() -> None:
train = SplitResult(
run_id="train",
trade_count=120,
profit_factor=1.6,
total_return_pct=18.0,
annualized_return_pct=18.0,
win_rate=0.52,
max_drawdown_pct=8.0,
sharpe_ratio=1.1,
avg_gross_exposure_pct=20.0,
avg_net_exposure_pct=20.0,
days_in_market_pct=25.0,
)
valid = SplitResult(
run_id="valid",
trade_count=50,
profit_factor=1.5,
total_return_pct=12.0,
annualized_return_pct=12.0,
win_rate=0.5,
max_drawdown_pct=7.0,
sharpe_ratio=1.0,
avg_gross_exposure_pct=20.0,
avg_net_exposure_pct=20.0,
days_in_market_pct=22.0,
)
test = SplitResult(
run_id="test",
trade_count=70,
profit_factor=1.3,
total_return_pct=8.0,
annualized_return_pct=8.0,
win_rate=0.48,
max_drawdown_pct=6.0,
sharpe_ratio=0.8,
avg_gross_exposure_pct=20.0,
avg_net_exposure_pct=20.0,
days_in_market_pct=20.0,
)
wf_summary = WalkForwardSummary(
train_days=252,
test_days=63,
step_days=63,
fold_count=4,
folds=[],
train_aggregate=WalkForwardAggregate(mean_return_pct=12.0, median_return_pct=11.0),
test_aggregate=WalkForwardAggregate(
mean_return_pct=9.0,
median_return_pct=8.0,
worst_return_pct=2.0,
positive_fold_rate_pct=75.0,
mean_profit_factor=1.4,
mean_max_drawdown_pct=7.0,
mean_trade_count=30.0,
mean_win_rate=0.5,
),
gap_stats=WalkForwardGapStats(
mean_train_test_return_gap_pct=25.0,
worst_train_test_return_gap_pct=35.0,
fold_return_cv=0.5,
),
engine_reliability_ratio=1.0,
)
orbqs, breakdown = compute_orbqs(
train,
valid,
test,
wf_summary,
{
"bear_2022": {"sharpe_ratio": -0.4, "max_drawdown_pct": 12.0},
"recovery_2023h1": {"sharpe_ratio": 0.9, "max_drawdown_pct": 9.0},
"bull_2023h2": {"sharpe_ratio": 1.2, "max_drawdown_pct": 8.0},
"oos_2026": {"sharpe_ratio": 0.8, "max_drawdown_pct": 6.0},
},
{
"is_oos": {"retention_pct": 75.0},
"walk_forward": {"mean_sharpe": 1.0, "cv": 0.5},
"param_plateau": {"params": [{"plateau": 0.8}]},
"permutation": {"p_value": 0.03},
},
)
assert orbqs is not None
assert 0.0 <= orbqs <= 100.0
assert breakdown["activity_factor"] == 0.85
assert "rqs_breakdown" in breakdown
assert "wfqs_v2_breakdown" in breakdown
assert "rrs_breakdown" in breakdown
assert "overfit_breakdown" in breakdown
@pytest.mark.asyncio
async def test_build_orb_research_context_reuses_snapshot(tmp_path, monkeypatch) -> None:
cache_dir = tmp_path / "intraday"
config = IntradayConfig(
strategy_mode="orb",
orb_strategy=ORBStrategyParams(
min_price=5.0,
min_atr_14=0.5,
min_avg_dollar_volume=1_000_000.0,
),
universe=UniverseParams(source="midlarge"),
backtest=BacktestParams(),
cache=CacheParams(enabled=True, dir=str(cache_dir)),
output=OutputParams(),
)
calls = {"fetch_daily": 0, "enrich": 0, "prescreen": 0}
async def fake_resolve_universe(*args, **kwargs):
return ["AAA", "BBB"]
async def fake_get_trading_days(*args, **kwargs):
return ["2024-01-02", "2024-01-03"]
async def fake_fetch_daily(*args, **kwargs):
calls["fetch_daily"] += 1
return {
"AAA": [{"date": "2024-01-02", "open": 10, "high": 11, "low": 9, "close": 10.5, "volume": 1000}],
"BBB": [{"date": "2024-01-02", "open": 20, "high": 21, "low": 19, "close": 20.5, "volume": 2000}],
}
def fake_enrich(*args, **kwargs):
calls["enrich"] += 1
return {
"AAA": {"2024-01-02": {"atr_14": 1.0}},
"BBB": {"2024-01-02": {"atr_14": 1.2}},
}
def fake_prescreen(*args, **kwargs):
calls["prescreen"] += 1
return {
"2024-01-02": ["AAA", "BBB"],
"2024-01-03": ["AAA"],
}
monkeypatch.setattr(orb_research, "resolve_universe", fake_resolve_universe)
monkeypatch.setattr(orb_research, "get_trading_days", fake_get_trading_days)
monkeypatch.setattr(orb_research, "fetch_daily_bars_bulk", fake_fetch_daily)
monkeypatch.setattr(orb_research, "enrich_daily_bars", fake_enrich)
monkeypatch.setattr(orb_research, "orb_pre_screen_candidates", fake_prescreen)
context = await build_orb_research_context(
config,
"2024-01-02",
"2024-01-03",
client=object(),
)
assert context.candidates["2024-01-02"] == ["AAA", "BBB"]
assert calls == {"fetch_daily": 1, "enrich": 1, "prescreen": 1}
snapshot_dir = cache_dir.with_name("orb_research")
assert any(snapshot_dir.rglob("*.pkl.gz"))
async def fail_fetch(*args, **kwargs):
raise AssertionError("daily fetch should not run on snapshot hit")
def fail_enrich(*args, **kwargs):
raise AssertionError("enrichment should not run on snapshot hit")
def fail_prescreen(*args, **kwargs):
raise AssertionError("pre-screen should not run on snapshot hit")
monkeypatch.setattr(orb_research, "fetch_daily_bars_bulk", fail_fetch)
monkeypatch.setattr(orb_research, "enrich_daily_bars", fail_enrich)
monkeypatch.setattr(orb_research, "orb_pre_screen_candidates", fail_prescreen)
cached_context = await build_orb_research_context(
config,
"2024-01-02",
"2024-01-03",
client=object(),
)
assert cached_context.tickers == ["AAA", "BBB"]
assert cached_context.trading_days == ["2024-01-02", "2024-01-03"]
assert cached_context.candidates == context.candidates
@pytest.mark.asyncio
async def test_simulate_orb_period_reuses_period_metrics_cache(tmp_path, monkeypatch) -> None:
cache_dir = tmp_path / "intraday"
config = IntradayConfig(
strategy_mode="orb",
orb_strategy=ORBStrategyParams(),
universe=UniverseParams(source="midlarge"),
backtest=BacktestParams(),
cache=CacheParams(enabled=True, dir=str(cache_dir)),
output=OutputParams(),
)
eval_cache = orb_research.ORBPeriodMetricsCache(cache_dir.with_name("orb_eval"))
context = orb_research.ORBResearchContext(
config=config,
tickers=["AAA"],
trading_days=["2024-01-02"],
daily_bars={"AAA": []},
enrichment={},
candidates={"2024-01-02": ["AAA"]},
cache=None,
daily_cache=None,
eval_cache=eval_cache,
tape_cache=None,
oracle_url="http://localhost:18001",
research_snapshot_key="snapshot_key",
)
calls = {"fetch": 0, "simulate": 0}
async def fake_fetch_intraday(*args, **kwargs):
calls["fetch"] += 1
return {"AAA": {"2024-01-02": [{"timestamp": "2024-01-02T09:35:00-05:00"}]}}
def fake_run_sim(*args, **kwargs):
calls["simulate"] += 1
return ([], None)
monkeypatch.setattr(orb_research, "fetch_intraday_bulk", fake_fetch_intraday)
monkeypatch.setattr(orb_research, "run_orb_simulation_with_state", fake_run_sim)
metrics_first = await orb_research.simulate_orb_period(
context,
client=object(),
orb_params=config.orb_strategy or ORBStrategyParams(),
trading_days=["2024-01-02"],
run_id="first",
)
assert calls == {"fetch": 1, "simulate": 1}
assert metrics_first.run_id == "first"
async def fail_fetch(*args, **kwargs):
raise AssertionError("intraday fetch should not run on period cache hit")
def fail_run(*args, **kwargs):
raise AssertionError("simulation should not run on period cache hit")
monkeypatch.setattr(orb_research, "fetch_intraday_bulk", fail_fetch)
monkeypatch.setattr(orb_research, "run_orb_simulation_with_state", fail_run)
metrics_second = await orb_research.simulate_orb_period(
context,
client=object(),
orb_params=config.orb_strategy or ORBStrategyParams(),
trading_days=["2024-01-02"],
run_id="second",
)
assert metrics_second.run_id == "second"
assert metrics_second.trading_days == metrics_first.trading_days
@pytest.mark.asyncio
async def test_simulate_orb_period_resumes_from_chunk_checkpoint(tmp_path, monkeypatch) -> None:
cache_dir = tmp_path / "intraday"
config = IntradayConfig(
strategy_mode="orb",
orb_strategy=ORBStrategyParams(initial_capital=10_000.0),
universe=UniverseParams(source="midlarge"),
backtest=BacktestParams(),
cache=CacheParams(enabled=True, dir=str(cache_dir)),
output=OutputParams(),
)
eval_cache = orb_research.ORBPeriodMetricsCache(cache_dir.with_name("orb_eval"))
context = orb_research.ORBResearchContext(
config=config,
tickers=["AAA"],
trading_days=["2024-01-02", "2024-01-03", "2024-01-04"],
daily_bars={"AAA": []},
enrichment={},
candidates={
"2024-01-02": ["AAA"],
"2024-01-03": ["AAA"],
"2024-01-04": ["AAA"],
},
cache=None,
daily_cache=None,
eval_cache=eval_cache,
tape_cache=None,
oracle_url="http://localhost:18001",
research_snapshot_key="snapshot_key",
)
fetch_calls: list[str] = []
run_states: list[float | None] = []
first_run = {"attempt": True}
async def flaky_fetch(chunk_candidates, *args, **kwargs):
day = next(iter(chunk_candidates))
fetch_calls.append(day)
if first_run["attempt"] and day == "2024-01-03":
raise RuntimeError("oracle timeout")
return {
day: {
"AAA": [
{
"timestamp": f"{day}T09:35:00-05:00",
"open": 100.0,
"high": 101.0,
"low": 99.5,
"close": 100.5,
"volume": 1000.0,
}
]
}
}
def fake_run_sim(
all_intraday,
trading_days,
params,
enrichment,
ticker_sectors=None,
state=None,
progress_callback=None,
**kwargs,
):
run_states.append(state.equity if state is not None else None)
day_results = [
DayResult(date=day, daily_pnl=10.0, daily_return_pct=0.001)
for day in trading_days
]
next_equity = (state.equity if state is not None else params.initial_capital) + 10.0 * len(trading_days)
return (
day_results,
ORBSimulationState(
equity=next_equity,
ticker_last_traded={"AAA": trading_days[-1]},
settled_cash=None,
pending_settlements=[],
),
)
monkeypatch.setattr(orb_research, "fetch_intraday_bulk", flaky_fetch)
monkeypatch.setattr(orb_research, "run_orb_simulation_with_state", fake_run_sim)
with pytest.raises(RuntimeError, match="oracle timeout"):
await orb_research.simulate_orb_period(
context,
client=object(),
orb_params=config.orb_strategy or ORBStrategyParams(),
trading_days=context.trading_days,
run_id="resume-test",
max_pairs_per_chunk=1,
)
cache_key = eval_cache.build_key(
research_snapshot_key="snapshot_key",
orb_params=config.orb_strategy or ORBStrategyParams(),
trading_days=context.trading_days,
shuffle_candidates_seed=None,
)
checkpoint = eval_cache.load_checkpoint(cache_key)
assert checkpoint is not None
assert checkpoint["completed_chunks"] == 1
first_run["attempt"] = False
metrics = await orb_research.simulate_orb_period(
context,
client=object(),
orb_params=config.orb_strategy or ORBStrategyParams(),
trading_days=context.trading_days,
run_id="resume-test",
max_pairs_per_chunk=1,
)
assert fetch_calls == ["2024-01-02", "2024-01-03", "2024-01-03", "2024-01-04"]
assert run_states == [None, 10010.0, 10020.0]
assert metrics.trading_days == 3
assert metrics.final_equity == 10030.0
assert eval_cache.load(cache_key) is not None
assert eval_cache.load_checkpoint(cache_key) is None
@pytest.mark.asyncio
async def test_simulate_orb_period_reuses_prepared_tape_for_new_params(tmp_path, monkeypatch) -> None:
cache_dir = tmp_path / "intraday"
config = IntradayConfig(
strategy_mode="orb",
orb_strategy=ORBStrategyParams(),
universe=UniverseParams(source="midlarge"),
backtest=BacktestParams(),
cache=CacheParams(enabled=True, dir=str(cache_dir)),
output=OutputParams(),
)
tape_cache = orb_research.ORBPreparedTapeStore(cache_dir.with_name("orb_tape"))
context = orb_research.ORBResearchContext(
config=config,
tickers=["AAA"],
trading_days=["2024-01-02"],
daily_bars={"AAA": []},
enrichment={},
candidates={"2024-01-02": ["AAA"]},
cache=None,
daily_cache=None,
eval_cache=None,
tape_cache=tape_cache,
oracle_url="http://localhost:18001",
research_snapshot_key="snapshot_key",
)
calls = {"fetch": 0, "simulate": 0}
async def fake_fetch_intraday(*args, **kwargs):
calls["fetch"] += 1
return {
"2024-01-02": {
"AAA": [
{
"timestamp": "2024-01-02T09:35:00-05:00",
"open": 100.0,
"high": 101.0,
"low": 99.5,
"close": 100.5,
"volume": 1000.0,
}
]
}
}
def fake_run_sim(*args, **kwargs):
calls["simulate"] += 1
return ([DayResult(date="2024-01-02", daily_pnl=0.0, daily_return_pct=0.0)], ORBSimulationState(equity=10_000.0))
monkeypatch.setattr(orb_research, "fetch_intraday_bulk", fake_fetch_intraday)
monkeypatch.setattr(orb_research, "run_orb_simulation_with_state", fake_run_sim)
await orb_research.simulate_orb_period(
context,
client=object(),
orb_params=ORBStrategyParams(atr_stop_multiplier=1.0),
trading_days=["2024-01-02"],
run_id="tape-first",
)
assert calls == {"fetch": 1, "simulate": 1}
assert any((cache_dir.with_name("orb_tape")).rglob("*.pkl.gz"))
async def fail_fetch(*args, **kwargs):
raise AssertionError("raw intraday fetch should not run on tape hit")
monkeypatch.setattr(orb_research, "fetch_intraday_bulk", fail_fetch)
await orb_research.simulate_orb_period(
context,
client=object(),
orb_params=ORBStrategyParams(atr_stop_multiplier=1.25),
trading_days=["2024-01-02"],
run_id="tape-second",
)
assert calls == {"fetch": 1, "simulate": 2}

182
tests/unit/orb_trader/test_record_trade.py
Unescape Escape View File

@ -0,0 +1,182 @@
"""Unit tests for ORBTradingEngine._record_trade and run_post_close."""
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import MagicMock, call
import pytest
from apps.orb_trader.engine import ORBTradingEngine
from apps.orb_trader.models import ORBPositionRow
def _make_position(
*,
ticker: str = "AAPL",
direction: str = "long",
entry_price: float = 100.0,
shares: int = 10,
stop_distance: float = 2.0,
) -> ORBPositionRow:
return ORBPositionRow(
session_id="test-session",
date="2026-01-05",
ticker=ticker,
direction=direction,
entry_price=entry_price,
entry_time="2026-01-05T09:40:00",
shares=shares,
orb_high=entry_price * 1.01,
orb_low=entry_price * 0.99,
atr_at_entry=stop_distance / 0.75,
stop_distance=stop_distance,
current_stop=entry_price - stop_distance,
peak_price=entry_price,
rvol=2.0,
composite_score=0.7,
order_id="order-1",
)
def _make_engine() -> ORBTradingEngine:
params = SimpleNamespace(
daily_budget_reset=True,
drawdown_governor_threshold=None,
drawdown_governor_min_scale=0.30,
streak_sizing_win_bonus=None,
streak_sizing_loss_penalty=None,
streak_sizing_max=2.5,
streak_sizing_min=0.5,
)
session = SimpleNamespace(
session_id="test-session",
session_name="test",
initial_equity=10_000.0,
)
state = MagicMock()
state.get_equity.return_value = 10_000.0
state.get_peak_equity.return_value = 10_000.0
state.list_trades.return_value = []
engine = object.__new__(ORBTradingEngine)
engine._session = session
engine._params = params
engine._state = state
engine._log_callback = None
engine._date_str = "2026-01-05"
return engine
# ── _record_trade ─────────────────────────────────────────────────────────────
class TestRecordTrade:
def test_long_profit_pnl(self):
eng = _make_engine()
pos = _make_position(direction="long", entry_price=100.0, shares=10)
eng._record_trade(pos, exit_price=105.0, exit_time="T", exit_reason="close", equity=10_000.0)
saved = eng._state.save_trade.call_args[0][0]
assert saved.pnl == pytest.approx((105.0 - 100.0) * 10)
def test_long_loss_pnl(self):
eng = _make_engine()
pos = _make_position(direction="long", entry_price=100.0, shares=10)
eng._record_trade(pos, exit_price=96.0, exit_time="T", exit_reason="stop_loss", equity=10_000.0)
saved = eng._state.save_trade.call_args[0][0]
assert saved.pnl == pytest.approx((96.0 - 100.0) * 10) # -40.0
def test_short_profit_pnl(self):
# Short: profit = entry - exit (price falls)
eng = _make_engine()
pos = _make_position(direction="short", entry_price=100.0, shares=10)
eng._record_trade(pos, exit_price=92.0, exit_time="T", exit_reason="trailing_stop", equity=10_000.0)
saved = eng._state.save_trade.call_args[0][0]
assert saved.pnl == pytest.approx((100.0 - 92.0) * 10) # +80.0
def test_short_loss_pnl(self):
# Short: loss = entry - exit when price rises
eng = _make_engine()
pos = _make_position(direction="short", entry_price=100.0, shares=10)
eng._record_trade(pos, exit_price=104.0, exit_time="T", exit_reason="stop_loss", equity=10_000.0)
saved = eng._state.save_trade.call_args[0][0]
assert saved.pnl == pytest.approx((100.0 - 104.0) * 10) # -40.0
def test_r_multiple_long(self):
# entry=100, exit=106, stop_distance=2 → pnl=60, risk=20 → R=3.0
eng = _make_engine()
pos = _make_position(direction="long", entry_price=100.0, shares=10, stop_distance=2.0)
eng._record_trade(pos, exit_price=106.0, exit_time="T", exit_reason="close", equity=10_000.0)
saved = eng._state.save_trade.call_args[0][0]
assert saved.r_multiple == pytest.approx(3.0)
def test_position_closed_in_db(self):
eng = _make_engine()
pos = _make_position(direction="long", entry_price=100.0, shares=10)
eng._record_trade(pos, exit_price=105.0, exit_time="T", exit_reason="close", equity=10_000.0)
eng._state.close_position_record.assert_called_once_with(
"test-session", "2026-01-05", "AAPL"
)
def test_exit_reason_preserved(self):
for reason in ("close", "stop_loss", "trailing_stop"):
eng = _make_engine()
pos = _make_position(direction="long", entry_price=100.0, shares=5)
eng._record_trade(pos, exit_price=102.0, exit_time="T", exit_reason=reason, equity=10_000.0)
saved = eng._state.save_trade.call_args[0][0]
assert saved.exit_reason == reason
# ── run_post_close ────────────────────────────────────────────────────────────
class TestRunPostClose:
def test_equity_accumulates_daily_pnl(self):
eng = _make_engine()
eng._state.get_equity.return_value = 10_200.0 # prev equity
eng._state.list_trades.return_value = [
{"date": "2026-01-05", "pnl": 300.0, "exit_reason": "close"},
]
result = eng.run_post_close("2026-01-05")
assert result["equity"] == pytest.approx(10_500.0)
assert result["daily_pnl"] == pytest.approx(300.0)
def test_stops_hit_counts_stop_and_trailing(self):
eng = _make_engine()
eng._state.get_equity.return_value = 9_500.0
eng._state.list_trades.return_value = [
{"date": "2026-01-05", "pnl": -200.0, "exit_reason": "stop_loss"},
{"date": "2026-01-05", "pnl": 50.0, "exit_reason": "trailing_stop"},
{"date": "2026-01-05", "pnl": 400.0, "exit_reason": "close"},
]
result = eng.run_post_close("2026-01-05")
assert result["stops_hit"] == 2
assert result["trades"] == 3
def test_only_today_trades_counted(self):
eng = _make_engine()
eng._state.list_trades.return_value = [
{"date": "2026-01-04", "pnl": 500.0, "exit_reason": "close"}, # yesterday
{"date": "2026-01-05", "pnl": 100.0, "exit_reason": "close"}, # today
]
result = eng.run_post_close("2026-01-05")
assert result["daily_pnl"] == pytest.approx(100.0)
assert result["trades"] == 1
def test_snapshot_saved(self):
eng = _make_engine()
eng._state.list_trades.return_value = [
{"date": "2026-01-05", "pnl": 250.0, "exit_reason": "close"},
]
eng.run_post_close("2026-01-05")
eng._state.save_daily_snapshot.assert_called_once()
snap = eng._state.save_daily_snapshot.call_args[0][0]
assert snap.date == "2026-01-05"
assert snap.daily_pnl == pytest.approx(250.0)
Write Preview
Loading…
Cancel
Save