dennisthiessenandClaude Fable 5
565484de87
The run-id marker proved which scan wrote last, but the shadow book still selected setups by detected_at >= scan_start. An overlapping manual scan could insert rows in that same window; if the pipeline's scan wrote the marker last its id matched and the shadow book proceeded, then swept in -- or ranked highest -- a manual-scan row. The identity check gated entry but selection did not. Carry the run id onto the rows. Migration 025 adds an indexed trade_setups.scan_run_id. scan_all_tickers computes one id per run (pipeline's when a step, else fresh), passes it to scan_ticker which stamps every row after enhancement, and writes the same id to the completion marker. The shadow book selects WHERE scan_run_id == the matched id, so a concurrent scan's rows are excluded by identity regardless of their detected_at. The now-unused STARTED marker is dropped; COMPLETED (freshness) and RUN_ID (identity) remain. Decisive test: the pipeline's id matches, but a same-window manual row with a higher rank is present and is excluded -- only the pipeline's own row is traded. A time-window select would have swept it in and ranked it first. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
432 lines
16 KiB
Python
432 lines
16 KiB
Python
"""Shadow book selection, sizing and book isolation.
|
|
|
|
The shadow book only has evidentiary value if it selects what the backtest
|
|
would select: top-ranked qualified setups, up to capacity, skipping held names
|
|
and post-stop gate-reset lockouts. These tests pin that contract.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
from app.models.paper_trade import PaperTrade
|
|
from app.models.ticker import Ticker
|
|
from app.models.trade_setup import TradeSetup
|
|
from app.models.user import User
|
|
from app.services import shadow_book_service
|
|
from app.services.trade_policy import MANUAL_BOOK, SHADOW_BOOK, get_reentry_gate_locks
|
|
|
|
|
|
@pytest.fixture
|
|
async def session():
|
|
from tests.conftest import _test_session_factory
|
|
|
|
async with _test_session_factory() as session:
|
|
yield session
|
|
|
|
|
|
# Floors the gate applies; every setup below clears them so tests exercise
|
|
# ranking rather than qualification.
|
|
_CONFIG = {
|
|
"min_rr": 2.0,
|
|
"min_confidence": 0.0,
|
|
"min_momentum_percentile": 80.0,
|
|
"exclude_neutral": False,
|
|
}
|
|
|
|
|
|
async def _seed(session, symbols: list[str]) -> dict[str, int]:
|
|
session.add(User(id=1, username="owner", password_hash="x"))
|
|
ids: dict[str, int] = {}
|
|
for i, symbol in enumerate(symbols, start=1):
|
|
ticker = Ticker(id=i, symbol=symbol, name=symbol)
|
|
session.add(ticker)
|
|
ids[symbol] = i
|
|
await session.commit()
|
|
return ids
|
|
|
|
|
|
def _setup(
|
|
ticker_id: int,
|
|
*,
|
|
rank: float,
|
|
detected: datetime,
|
|
entry=100.0,
|
|
stop=95.0,
|
|
direction="long",
|
|
scan_run_id: str = "scan-run",
|
|
):
|
|
reward = abs(entry - stop) * 3
|
|
target = entry + reward if direction == "long" else entry - reward
|
|
return TradeSetup(
|
|
ticker_id=ticker_id,
|
|
direction=direction,
|
|
entry_price=entry,
|
|
stop_loss=stop,
|
|
target=target,
|
|
rr_ratio=3.0,
|
|
composite_score=70.0,
|
|
confidence_score=70.0,
|
|
detected_at=detected,
|
|
strategy_rank=rank,
|
|
momentum_percentile=90.0,
|
|
recommended_action="buy",
|
|
scan_run_id=scan_run_id,
|
|
targets_json=json.dumps(
|
|
[{"price": target, "probability": 45.0, "is_primary": True, "rr": 3.0}]
|
|
),
|
|
)
|
|
|
|
|
|
async def _mark_scan(
|
|
session,
|
|
*,
|
|
started: datetime | None = None,
|
|
completed: datetime | None = None,
|
|
run_id: str = "scan-run",
|
|
):
|
|
"""Record a successful scan run so the shadow book has something to act on.
|
|
|
|
``started`` is accepted for readability at call sites but only ``completed``
|
|
(freshness) and ``run_id`` (identity) are persisted.
|
|
"""
|
|
from app.services import rr_scanner_service as rr
|
|
|
|
completed = completed or started or datetime.now(timezone.utc)
|
|
await shadow_book_service.settings_store.upsert_setting(
|
|
session, rr.KEY_LAST_SCAN_COMPLETED, completed.isoformat()
|
|
)
|
|
await shadow_book_service.settings_store.upsert_setting(
|
|
session, rr.KEY_LAST_SCAN_RUN_ID, run_id
|
|
)
|
|
await session.commit()
|
|
|
|
|
|
class TestSizing:
|
|
def test_risks_one_percent_down_to_the_stop(self):
|
|
shares = shadow_book_service.position_shares(100_000, 1.0, 100.0, 95.0)
|
|
assert shares == pytest.approx(200.0) # $1,000 risk / $5 per share
|
|
|
|
def test_zero_risk_distance_takes_no_position(self):
|
|
assert shadow_book_service.position_shares(100_000, 1.0, 100.0, 100.0) == 0.0
|
|
|
|
def test_tight_stop_is_capped_at_the_notional_limit(self):
|
|
"""Without the cap, 1% risk on a $0.50 stop is a 2x-equity position."""
|
|
shares = shadow_book_service.position_shares(100_000, 1.0, 100.0, 99.5)
|
|
# Risk sizing alone wants 2,000 shares ($200k); the 20% cap allows 200.
|
|
assert shares == pytest.approx(200.0)
|
|
assert shares * 100.0 <= 100_000 * shadow_book_service.NOTIONAL_CAP
|
|
|
|
def test_cannot_spend_cash_it_does_not_have(self):
|
|
shares = shadow_book_service.position_shares(
|
|
100_000, 1.0, 100.0, 95.0, cash_available=5_000
|
|
)
|
|
assert shares == pytest.approx(50.0)
|
|
|
|
def test_no_cash_means_no_position(self):
|
|
assert (
|
|
shadow_book_service.position_shares(
|
|
100_000, 1.0, 100.0, 95.0, cash_available=0
|
|
)
|
|
== 0.0
|
|
)
|
|
|
|
|
|
class TestSelection:
|
|
@pytest.mark.asyncio
|
|
async def test_takes_top_ranked_up_to_capacity(self, session):
|
|
ids = await _seed(session, ["AAA", "BBB", "CCC"])
|
|
now = datetime.now(timezone.utc)
|
|
scan_start = now - timedelta(minutes=5)
|
|
session.add_all(
|
|
[
|
|
_setup(ids["AAA"], rank=0.10, detected=now),
|
|
_setup(ids["BBB"], rank=0.90, detected=now),
|
|
_setup(ids["CCC"], rank=0.50, detected=now),
|
|
]
|
|
)
|
|
await session.commit()
|
|
await _mark_scan(session, started=scan_start, completed=now)
|
|
await shadow_book_service.settings_store.upsert_setting(
|
|
session, shadow_book_service.KEY_CAPACITY, "2"
|
|
)
|
|
await session.commit()
|
|
|
|
summary = await shadow_book_service.open_shadow_positions(
|
|
session, activation_config=_CONFIG
|
|
)
|
|
|
|
assert summary["opened"] == 2
|
|
# Highest strategy_rank first — the backtest's ordering key.
|
|
assert summary["symbols"] == [ids["BBB"], ids["CCC"]]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skips_names_already_held(self, session):
|
|
ids = await _seed(session, ["AAA", "BBB"])
|
|
now = datetime.now(timezone.utc)
|
|
session.add_all(
|
|
[_setup(ids["AAA"], rank=0.9, detected=now), _setup(ids["BBB"], rank=0.5, detected=now)]
|
|
)
|
|
session.add(
|
|
PaperTrade(
|
|
user_id=1, ticker_id=ids["AAA"], direction="long", entry_price=100.0,
|
|
shares=10.0, stop_loss=95.0, target=115.0, status="open",
|
|
opened_at=now, book=SHADOW_BOOK,
|
|
)
|
|
)
|
|
await session.commit()
|
|
await _mark_scan(session, started=now - timedelta(minutes=5), completed=now)
|
|
|
|
summary = await shadow_book_service.open_shadow_positions(
|
|
session, activation_config=_CONFIG
|
|
)
|
|
|
|
assert summary["skipped_held"] == 1
|
|
assert summary["symbols"] == [ids["BBB"]]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respects_post_stop_gate_lock(self, session):
|
|
ids = await _seed(session, ["AAA"])
|
|
now = datetime.now(timezone.utc)
|
|
session.add(_setup(ids["AAA"], rank=0.9, detected=now))
|
|
# Stopped out and never requalified — locked out of re-entry.
|
|
session.add(
|
|
PaperTrade(
|
|
user_id=1, ticker_id=ids["AAA"], direction="long", entry_price=100.0,
|
|
shares=10.0, stop_loss=95.0, target=115.0, status="closed",
|
|
opened_at=now - timedelta(days=5), closed_at=now - timedelta(days=1),
|
|
close_price=95.0, close_reason="stop", book=SHADOW_BOOK,
|
|
)
|
|
)
|
|
await session.commit()
|
|
await _mark_scan(session, started=now - timedelta(minutes=5), completed=now)
|
|
|
|
summary = await shadow_book_service.open_shadow_positions(
|
|
session, activation_config=_CONFIG
|
|
)
|
|
|
|
assert summary["opened"] == 0
|
|
assert summary["skipped_locked"] == 1
|
|
|
|
|
|
class TestScanFreshness:
|
|
@pytest.mark.asyncio
|
|
async def test_no_scan_marker_means_no_trades(self, session):
|
|
"""A fresh DB / never-run scan must not trade anything."""
|
|
ids = await _seed(session, ["AAA"])
|
|
session.add(_setup(ids["AAA"], rank=0.9, detected=datetime.now(timezone.utc)))
|
|
await session.commit()
|
|
|
|
summary = await shadow_book_service.open_shadow_positions(
|
|
session, activation_config=_CONFIG
|
|
)
|
|
|
|
assert summary["opened"] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stale_scan_marker_refuses_even_fresh_looking_setups(self, session):
|
|
"""If this pipeline's scan failed/was disabled, the completion marker is
|
|
from a prior session — refuse, no matter how recent the setup rows look."""
|
|
ids = await _seed(session, ["AAA"])
|
|
now = datetime.now(timezone.utc)
|
|
session.add(_setup(ids["AAA"], rank=0.9, detected=now))
|
|
await session.commit()
|
|
# Marker is a day old → no scan ran in this pass.
|
|
await _mark_scan(session, started=now - timedelta(days=1, minutes=5),
|
|
completed=now - timedelta(days=1))
|
|
|
|
summary = await shadow_book_service.open_shadow_positions(
|
|
session, activation_config=_CONFIG
|
|
)
|
|
|
|
assert summary["opened"] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_setups_from_another_run_are_excluded_by_identity(self, session):
|
|
"""A row from a different scan run must not be traded even if its
|
|
detected_at falls in the same window — selection is by scan_run_id."""
|
|
ids = await _seed(session, ["AAA", "BBB"])
|
|
now = datetime.now(timezone.utc)
|
|
session.add_all(
|
|
[
|
|
_setup(ids["AAA"], rank=0.9, detected=now, scan_run_id="scan-run"),
|
|
# Same time window, different run — e.g. an overlapping manual scan.
|
|
_setup(ids["BBB"], rank=0.8, detected=now, scan_run_id="other-run"),
|
|
]
|
|
)
|
|
await session.commit()
|
|
await _mark_scan(session, completed=now, run_id="scan-run")
|
|
|
|
summary = await shadow_book_service.open_shadow_positions(
|
|
session, activation_config=_CONFIG
|
|
)
|
|
|
|
assert summary["symbols"] == [ids["AAA"]]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_newer_unqualified_row_suppresses_older_qualified(self, session):
|
|
"""Dedup happens before qualification: a fresh unqualified row for a
|
|
ticker must beat an earlier qualified row, not the other way round."""
|
|
ids = await _seed(session, ["AAA"])
|
|
now = datetime.now(timezone.utc)
|
|
# Earlier row qualifies; later row fails the R:R floor (rr 1.0 < 2.0).
|
|
# Both belong to the same scan run.
|
|
older = _setup(ids["AAA"], rank=0.9, detected=now - timedelta(minutes=8))
|
|
newer = TradeSetup(
|
|
ticker_id=ids["AAA"], direction="long", entry_price=100.0,
|
|
stop_loss=95.0, target=105.0, rr_ratio=1.0, composite_score=70.0,
|
|
confidence_score=70.0, detected_at=now, strategy_rank=0.9,
|
|
momentum_percentile=90.0, recommended_action="buy",
|
|
scan_run_id="scan-run",
|
|
targets_json=json.dumps(
|
|
[{"price": 105.0, "probability": 45.0, "is_primary": True, "rr": 1.0}]
|
|
),
|
|
)
|
|
session.add_all([older, newer])
|
|
await session.commit()
|
|
await _mark_scan(session, completed=now, run_id="scan-run")
|
|
|
|
summary = await shadow_book_service.open_shadow_positions(
|
|
session, activation_config=_CONFIG
|
|
)
|
|
|
|
assert summary["opened"] == 0
|
|
|
|
|
|
class TestPipelineScanBinding:
|
|
@pytest.mark.asyncio
|
|
async def test_pipeline_scan_run_id_match_is_accepted(self, session):
|
|
"""The scan that stamped this pipeline's run id is the one to act on."""
|
|
ids = await _seed(session, ["AAA"])
|
|
now = datetime.now(timezone.utc)
|
|
session.add(_setup(ids["AAA"], rank=0.9, detected=now, scan_run_id="pipeline-A"))
|
|
await session.commit()
|
|
await _mark_scan(session, completed=now, run_id="pipeline-A")
|
|
|
|
summary = await shadow_book_service.open_shadow_positions(
|
|
session, activation_config=_CONFIG, expected_run_id="pipeline-A"
|
|
)
|
|
|
|
assert summary["opened"] == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_manual_row_excluded_even_when_pipeline_matches(self, session):
|
|
"""The reported P2: the pipeline's scan matches (it wrote the marker
|
|
last), but an overlapping manual scan inserted a row in the same time
|
|
window. Identity selection must exclude that manual row — a detected_at
|
|
window would have swept it in."""
|
|
ids = await _seed(session, ["AAA", "BBB"])
|
|
now = datetime.now(timezone.utc)
|
|
session.add_all(
|
|
[
|
|
_setup(ids["AAA"], rank=0.9, detected=now, scan_run_id="pipeline-A"),
|
|
# Overlapping manual scan, same window, higher rank — must NOT win.
|
|
_setup(ids["BBB"], rank=0.99, detected=now, scan_run_id="manual-X"),
|
|
]
|
|
)
|
|
await session.commit()
|
|
await _mark_scan(session, completed=now, run_id="pipeline-A")
|
|
|
|
summary = await shadow_book_service.open_shadow_positions(
|
|
session, activation_config=_CONFIG, expected_run_id="pipeline-A"
|
|
)
|
|
|
|
assert summary["symbols"] == [ids["AAA"]]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_manual_scan_finishing_last_is_refused(self, session):
|
|
"""A manual rr_scanner overlaps the pipeline and writes the markers last.
|
|
Its completion timestamp is later than the pipeline start, but its run id
|
|
is not the pipeline's — refuse, though a timestamp check would accept."""
|
|
ids = await _seed(session, ["AAA"])
|
|
now = datetime.now(timezone.utc)
|
|
session.add(_setup(ids["AAA"], rank=0.9, detected=now, scan_run_id="manual-999"))
|
|
await session.commit()
|
|
# Pipeline expects "pipeline-A"; the manual scan's id won the last write.
|
|
await _mark_scan(session, completed=now, run_id="manual-999")
|
|
|
|
summary = await shadow_book_service.open_shadow_positions(
|
|
session, activation_config=_CONFIG, expected_run_id="pipeline-A"
|
|
)
|
|
|
|
assert summary["opened"] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pipeline_scan_failed_leaves_prior_run_id(self, session):
|
|
"""If the pipeline's own scan failed, the stored id is a prior run's."""
|
|
ids = await _seed(session, ["AAA"])
|
|
now = datetime.now(timezone.utc)
|
|
session.add(_setup(ids["AAA"], rank=0.9, detected=now, scan_run_id="yesterday"))
|
|
await session.commit()
|
|
await _mark_scan(session, completed=now, run_id="yesterday")
|
|
|
|
summary = await shadow_book_service.open_shadow_positions(
|
|
session, activation_config=_CONFIG, expected_run_id="pipeline-today"
|
|
)
|
|
|
|
assert summary["opened"] == 0
|
|
|
|
|
|
class TestLongOnly:
|
|
@pytest.mark.asyncio
|
|
async def test_shorts_are_never_taken_even_with_gate_disabled(self, session):
|
|
"""min_momentum_percentile=0 lets shorts pass the gate; shadow is always
|
|
long-only regardless, and its cash accounting assumes longs."""
|
|
ids = await _seed(session, ["AAA"])
|
|
now = datetime.now(timezone.utc)
|
|
session.add(
|
|
_setup(ids["AAA"], rank=0.9, detected=now, direction="short",
|
|
entry=100.0, stop=105.0)
|
|
)
|
|
await session.commit()
|
|
await _mark_scan(session, started=now - timedelta(minutes=5), completed=now)
|
|
|
|
gate_off = {**_CONFIG, "min_momentum_percentile": 0.0}
|
|
summary = await shadow_book_service.open_shadow_positions(
|
|
session, activation_config=gate_off
|
|
)
|
|
|
|
assert summary["opened"] == 0
|
|
|
|
|
|
class TestBookIsolation:
|
|
@pytest.mark.asyncio
|
|
async def test_gate_locks_do_not_leak_between_books(self, session):
|
|
"""A manual stop must not lock the shadow book out of the same name."""
|
|
ids = await _seed(session, ["AAA"])
|
|
now = datetime.now(timezone.utc)
|
|
session.add(
|
|
PaperTrade(
|
|
user_id=1, ticker_id=ids["AAA"], direction="long", entry_price=100.0,
|
|
shares=10.0, stop_loss=95.0, target=115.0, status="closed",
|
|
opened_at=now - timedelta(days=5), closed_at=now - timedelta(days=1),
|
|
close_price=95.0, close_reason="stop", book=MANUAL_BOOK,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
assert ids["AAA"] in await get_reentry_gate_locks(session, book=MANUAL_BOOK)
|
|
assert ids["AAA"] not in await get_reentry_gate_locks(session, book=SHADOW_BOOK)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shadow_equity_ignores_manual_pnl(self, session):
|
|
ids = await _seed(session, ["AAA"])
|
|
now = datetime.now(timezone.utc)
|
|
session.add(
|
|
PaperTrade(
|
|
user_id=1, ticker_id=ids["AAA"], direction="long", entry_price=100.0,
|
|
shares=100.0, stop_loss=95.0, target=115.0, status="closed",
|
|
opened_at=now - timedelta(days=5), closed_at=now,
|
|
close_price=150.0, close_reason="trailing", book=MANUAL_BOOK,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
equity, cash = await shadow_book_service.equity_and_cash(session, 100_000.0, [])
|
|
|
|
assert equity == 100_000.0
|
|
assert cash == 100_000.0
|