dennisthiessen
2311999e57
Log the 21:14 orphan as a snapshot-build race, rewrite the context table to authoritative ICs only, and soften the vol-tilt warning. Add extender completion manifest + breadth refuse guard; strip intermediate/orphaned reports; park the thread (no book sim, no deploy).
134 lines
4.4 KiB
Python
134 lines
4.4 KiB
Python
"""Completion-manifest guard for research.sqlite breadth runs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine, text
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SCRIPTS = ROOT / "scripts"
|
|
if str(SCRIPTS) not in sys.path:
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
|
|
from research_snapshot_manifest import ( # noqa: E402
|
|
assert_research_snapshot_complete,
|
|
clear_manifest,
|
|
load_manifest,
|
|
manifest_path_for,
|
|
write_completion_manifest,
|
|
)
|
|
|
|
|
|
def _tiny_research_db(path: Path, *, tickers: int = 3, bars_each: int = 5) -> None:
|
|
engine = create_engine(f"sqlite:///{path.resolve().as_posix()}", future=True)
|
|
with engine.begin() as conn:
|
|
conn.execute(
|
|
text(
|
|
"CREATE TABLE tickers ("
|
|
"id INTEGER PRIMARY KEY, symbol TEXT NOT NULL UNIQUE, "
|
|
"name TEXT, created_at TEXT)"
|
|
)
|
|
)
|
|
conn.execute(
|
|
text(
|
|
"CREATE TABLE ohlcv_records ("
|
|
"id INTEGER PRIMARY KEY, ticker_id INTEGER, date TEXT, "
|
|
"open REAL, high REAL, low REAL, close REAL, volume INTEGER, "
|
|
"created_at TEXT)"
|
|
)
|
|
)
|
|
conn.execute(
|
|
text(
|
|
"CREATE TABLE research_rank_only ("
|
|
"ticker_id INTEGER PRIMARY KEY, symbol TEXT NOT NULL UNIQUE)"
|
|
)
|
|
)
|
|
for i in range(tickers):
|
|
sym = f"T{i}"
|
|
conn.execute(
|
|
text(
|
|
"INSERT INTO tickers (id, symbol, name, created_at) "
|
|
"VALUES (:id, :sym, NULL, '2026-01-01')"
|
|
),
|
|
{"id": i + 1, "sym": sym},
|
|
)
|
|
if i > 0:
|
|
conn.execute(
|
|
text(
|
|
"INSERT INTO research_rank_only (ticker_id, symbol) "
|
|
"VALUES (:id, :sym)"
|
|
),
|
|
{"id": i + 1, "sym": sym},
|
|
)
|
|
for d in range(bars_each):
|
|
conn.execute(
|
|
text(
|
|
"INSERT INTO ohlcv_records "
|
|
"(ticker_id, date, open, high, low, close, volume, created_at) "
|
|
"VALUES (:tid, :date, 1,1,1,1,100, '2026-01-01')"
|
|
),
|
|
{"tid": i + 1, "date": f"2026-01-{d+1:02d}"},
|
|
)
|
|
engine.dispose()
|
|
|
|
|
|
def test_write_and_assert_complete(tmp_path: Path) -> None:
|
|
snap = tmp_path / "research.sqlite"
|
|
_tiny_research_db(snap)
|
|
path = write_completion_manifest(snap, complete=True, sources={"t": "unit"})
|
|
assert path == manifest_path_for(snap)
|
|
assert path.exists()
|
|
|
|
m = assert_research_snapshot_complete(snap)
|
|
assert m["complete"] is True
|
|
assert m["ticker_count"] == 3
|
|
assert m["ohlcv_row_count"] == 15
|
|
assert m["rank_only_count"] == 2
|
|
assert m["live_counts"]["ticker_count"] == 3
|
|
|
|
|
|
def test_refuse_missing_manifest(tmp_path: Path) -> None:
|
|
snap = tmp_path / "research.sqlite"
|
|
_tiny_research_db(snap)
|
|
with pytest.raises(SystemExit, match="manifest missing"):
|
|
assert_research_snapshot_complete(snap)
|
|
|
|
|
|
def test_refuse_incomplete_flag(tmp_path: Path) -> None:
|
|
snap = tmp_path / "research.sqlite"
|
|
_tiny_research_db(snap)
|
|
write_completion_manifest(snap, complete=False, limit=50)
|
|
with pytest.raises(SystemExit, match="marked incomplete"):
|
|
assert_research_snapshot_complete(snap)
|
|
|
|
|
|
def test_refuse_count_mismatch(tmp_path: Path) -> None:
|
|
snap = tmp_path / "research.sqlite"
|
|
_tiny_research_db(snap)
|
|
write_completion_manifest(snap, complete=True)
|
|
# Tamper: change live DB after manifest written
|
|
engine = create_engine(f"sqlite:///{snap.resolve().as_posix()}", future=True)
|
|
with engine.begin() as conn:
|
|
conn.execute(
|
|
text(
|
|
"INSERT INTO tickers (id, symbol, name, created_at) "
|
|
"VALUES (99, 'EXTRA', NULL, '2026-01-01')"
|
|
)
|
|
)
|
|
engine.dispose()
|
|
with pytest.raises(SystemExit, match="does not match"):
|
|
assert_research_snapshot_complete(snap)
|
|
|
|
|
|
def test_clear_manifest(tmp_path: Path) -> None:
|
|
snap = tmp_path / "research.sqlite"
|
|
_tiny_research_db(snap)
|
|
write_completion_manifest(snap, complete=True)
|
|
assert load_manifest(snap) is not None
|
|
clear_manifest(snap)
|
|
assert load_manifest(snap) is None
|