Compare commits

...
10 Commits
Author SHA1 Message Date
dennisthiessenandClaude Opus 5 d29d603158 test: drop redundant scanner primary-target suites
Deploy / lint (push) Successful in 29s
Deploy / test (push) Successful in 1m23s
Deploy / deploy (push) Successful in 42s
test_rr_scanner_bug_exploration.py and test_rr_scanner_fix_check.py both
assert one invariant: the headline target is the probability-based near
level, not the far max-R:R lottery. That is already covered directly by
test_recommendation_service.py's _select_primary_target tests, which also
reach cases these never did (empty list, probability floor, activation vs
scanner floor), and end to end by test_rr_scanner_integration.py's
full-flow test -- a strict superset of their deterministic cases: three
resistance and three support levels, both directions, plus persistence
and rr_ratio consistency.

The two files also duplicated each other, and their docstrings had gone
stale: test_deterministic_long_three_levels documented a hand-computed
_compute_quality_score winner even though the assertion is about the
probability primary that supersedes it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-04 10:11:26 +02:00
dennisthiessenandClaude Opus 5 70157ccfc2 perf: reuse the test schema instead of rebuilding it per test
The autouse _setup_db fixture ran create_all + drop_all for every test in
the suite, including the many that never open a session. That cycle costs
~49ms against these 22 tables; truncating them instead costs ~6ms for the
same guarantee of an empty database per test.

Build the schema once, then delete every row before each subsequent test.
No model sets sqlite_autoincrement, so SQLite reuses rowids after a full
delete and generated ids still restart at 1.

Measured over 874 tests, deterministic order: 138.6s -> 74.5s (~46%).
Verified green under pytest-randomly's default random ordering as well.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-04 10:11:20 +02:00
dennisthiessen f49b422095 fix: close remaining ingestion review gaps 2026-08-04 08:55:30 +02:00
dennisthiessen 59ac108c90 perf: scope SEC ticker quality checks 2026-08-04 08:09:37 +02:00
dennisthiessen e0f3d43efb fix: tighten max-hold session countdown 2026-08-04 08:07:10 +02:00
dennisthiessen 4c0c0579f5 fix: make SEC quality gating terminal-safe 2026-08-04 07:58:18 +02:00
dennisthiessen d1caac86b5 fix: preserve OHLCV stale detection 2026-08-04 07:39:40 +02:00
dennisthiessen d431ee283d feat: show max-hold session countdown 2026-08-03 23:55:30 +02:00
dennisthiessen 3a6900d45a fix: gate setups on SEC filing completeness 2026-08-03 23:47:07 +02:00
dennisthiessen 7d703ea524 fix: refresh same-day OHLCV bars 2026-08-03 23:13:17 +02:00
27 changed files with 1712 additions and 740 deletions
@@ -0,0 +1,147 @@
"""SEC filing retry queue and setup-quality gate
Revision ID: 028
Revises: 027
Create Date: 2026-08-03 00:00:00.000000
"""
from datetime import date, datetime, timezone
import json
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = "028"
down_revision: Union[str, None] = "027"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
"sec_filing_gaps",
sa.Column("id", sa.Integer(), primary_key=True),
sa.Column("cik", sa.String(length=10), nullable=False),
sa.Column("accession", sa.String(length=25), nullable=False),
sa.Column("form", sa.String(length=12), nullable=True),
sa.Column("index_date", sa.Date(), nullable=True),
sa.Column("reason", sa.String(length=64), nullable=False),
sa.Column("coregistrant_ciks_json", sa.Text(), nullable=True),
sa.Column("first_seen_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("last_attempted_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("escalated_at", sa.DateTime(timezone=True), nullable=True),
sa.UniqueConstraint("accession", name="uq_sec_filing_gaps_accession"),
)
op.create_index("ix_sec_filing_gaps_cik", "sec_filing_gaps", ["cik"])
_backfill_retry_queue()
def downgrade() -> None:
op.drop_index("ix_sec_filing_gaps_cik", table_name="sec_filing_gaps")
op.drop_table("sec_filing_gaps")
def _as_date(value) -> date | None:
if isinstance(value, datetime):
return value.date()
if isinstance(value, date):
return value
if isinstance(value, str):
try:
return date.fromisoformat(value)
except ValueError:
return None
return None
def _backfill_retry_queue() -> None:
"""Materialize pre-queue promoted gaps once; runtime never scans history."""
bind = op.get_bind()
runs = sa.table(
"data_import_runs",
sa.column("source", sa.String()),
sa.column("status", sa.String()),
sa.column("validation_json", sa.Text()),
sa.column("source_max_date", sa.Date()),
sa.column("started_at", sa.DateTime(timezone=True)),
)
snapshots = sa.table(
"fundamental_snapshots",
sa.column("cik", sa.String()),
sa.column("accession", sa.String()),
sa.column("filed_date", sa.Date()),
)
gaps = sa.table(
"sec_filing_gaps",
sa.column("cik", sa.String()),
sa.column("accession", sa.String()),
sa.column("form", sa.String()),
sa.column("index_date", sa.Date()),
sa.column("reason", sa.String()),
sa.column("coregistrant_ciks_json", sa.Text()),
sa.column("first_seen_at", sa.DateTime(timezone=True)),
sa.column("last_attempted_at", sa.DateTime(timezone=True)),
sa.column("escalated_at", sa.DateTime(timezone=True)),
)
snapshot_rows = bind.execute(
sa.select(snapshots.c.cik, snapshots.c.accession, snapshots.c.filed_date)
).all()
resolved_accessions = {row.accession for row in snapshot_rows}
latest_filed_by_cik: dict[str, date] = {}
for row in snapshot_rows:
if row.filed_date is not None:
current = latest_filed_by_cik.get(row.cik)
if current is None or row.filed_date > current:
latest_filed_by_cik[row.cik] = row.filed_date
audit_rows = bind.execute(
sa.select(
runs.c.validation_json,
runs.c.source_max_date,
runs.c.started_at,
).where(
runs.c.source == "sec_facts",
runs.c.status == "promoted",
runs.c.validation_json.is_not(None),
)
).all()
now = datetime.now(timezone.utc)
candidates: dict[str, dict] = {}
for audit in audit_rows:
try:
summary = json.loads(audit.validation_json)
except (TypeError, ValueError):
continue
if not isinstance(summary, dict):
continue
for item in summary.get("missing_xbrl") or []:
accession = item.get("accession")
raw_cik = item.get("cik")
if not accession or raw_cik is None or accession in resolved_accessions:
continue
cik = str(raw_cik).zfill(10)
index_date = _as_date(item.get("index_date")) or _as_date(
audit.source_max_date
)
later_filed = latest_filed_by_cik.get(cik)
if index_date is not None and later_filed is not None and later_filed > index_date:
continue
first_seen = audit.started_at or now
existing = candidates.get(accession)
if existing is not None and existing["first_seen_at"] <= first_seen:
continue
candidates[accession] = {
"cik": cik,
"accession": accession,
"form": item.get("form"),
"index_date": index_date,
"reason": item.get("reason") or "not_in_companyfacts",
"coregistrant_ciks_json": json.dumps(item.get("coregistrants") or []),
"first_seen_at": first_seen,
"last_attempted_at": first_seen,
"escalated_at": None,
}
if candidates:
op.bulk_insert(gaps, list(candidates.values()))
+2
View File
@@ -17,6 +17,7 @@ from app.models.regime_snapshot import RegimeSnapshot
from app.models.benchmark_price import BenchmarkPrice from app.models.benchmark_price import BenchmarkPrice
from app.models.signal_context_snapshot import SignalContextSnapshot from app.models.signal_context_snapshot import SignalContextSnapshot
from app.models.system_event import SystemEvent from app.models.system_event import SystemEvent
from app.models.sec_filing_gap import SecFilingGap
__all__ = [ __all__ = [
"Ticker", "Ticker",
@@ -40,4 +41,5 @@ __all__ = [
"BenchmarkPrice", "BenchmarkPrice",
"SignalContextSnapshot", "SignalContextSnapshot",
"SystemEvent", "SystemEvent",
"SecFilingGap",
] ]
+32
View File
@@ -0,0 +1,32 @@
from datetime import date, datetime
from sqlalchemy import Date, DateTime, Index, String, Text, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from app.database import Base
class SecFilingGap(Base):
"""Active SEC filing that could not yet be reconstructed.
Rows form a small retry queue. Successful snapshot ingestion deletes the
matching row; a later valid filing supersedes it. While a current row remains,
tickers mapped to its CIK are not eligible for actionable trade setups.
"""
__tablename__ = "sec_filing_gaps"
__table_args__ = (
UniqueConstraint("accession", name="uq_sec_filing_gaps_accession"),
Index("ix_sec_filing_gaps_cik", "cik"),
)
id: Mapped[int] = mapped_column(primary_key=True)
cik: Mapped[str] = mapped_column(String(10), nullable=False)
accession: Mapped[str] = mapped_column(String(25), nullable=False)
form: Mapped[str | None] = mapped_column(String(12), nullable=True)
index_date: Mapped[date | None] = mapped_column(Date, nullable=True)
reason: Mapped[str] = mapped_column(String(64), nullable=False)
coregistrant_ciks_json: Mapped[str | None] = mapped_column(Text, nullable=True)
first_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
last_attempted_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
escalated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+10 -1
View File
@@ -10,6 +10,7 @@ from app.schemas.common import APIEnvelope
from app.schemas.fundamental import FundamentalResponse from app.schemas.fundamental import FundamentalResponse
from app.services.fundamental_service import get_fundamental from app.services.fundamental_service import get_fundamental
from app.services.fundamentals_api_service import build_fundamentals_v1 from app.services.fundamentals_api_service import build_fundamentals_v1
from app.services import fundamentals_quality_service
router = APIRouter(tags=["fundamentals"]) router = APIRouter(tags=["fundamentals"])
@@ -34,6 +35,7 @@ async def read_fundamentals(
"""Get latest fundamental data for a symbol (legacy fields + additive v1).""" """Get latest fundamental data for a symbol (legacy fields + additive v1)."""
record = await get_fundamental(db, symbol) record = await get_fundamental(db, symbol)
v1 = await build_fundamentals_v1(db, symbol) v1 = await build_fundamentals_v1(db, symbol)
quality = await fundamentals_quality_service.ticker_quality(db, symbol)
legacy: dict = {} legacy: dict = {}
if record is not None: if record is not None:
@@ -47,5 +49,12 @@ async def read_fundamentals(
unavailable_fields=_parse_unavailable_fields(record.unavailable_fields_json), unavailable_fields=_parse_unavailable_fields(record.unavailable_fields_json),
) )
data = FundamentalResponse(symbol=symbol.strip().upper(), **legacy, **v1) data = FundamentalResponse(
symbol=symbol.strip().upper(),
setup_eligible=quality.eligible,
setup_block_code=quality.code,
setup_block_reason=quality.message,
**legacy,
**v1,
)
return APIEnvelope(status="success", data=data.model_dump()) return APIEnvelope(status="success", data=data.model_dump())
+11 -4
View File
@@ -512,13 +512,14 @@ async def collect_ohlcv(
job_name: str = "data_collector", job_name: str = "data_collector",
*, *,
refetch_days: int = 0, refetch_days: int = 0,
refresh_sr: bool = True,
) -> None: ) -> None:
"""Fetch latest daily OHLCV for all tracked tickers. """Fetch latest daily OHLCV for all tracked tickers.
Uses AlpacaOHLCVProvider. Processes each ticker independently. Uses AlpacaOHLCVProvider. Processes each ticker independently.
On rate limit, records last successful ticker for resume. On rate limit, records last successful ticker for resume.
Start date is resolved by ingestion progress: Start date is resolved by ingestion progress:
- existing ticker: resume from last_ingested_date + 1 - existing ticker: overlap last_ingested_date so partial bars refresh
- new ticker: backfill the configured history window - new ticker: backfill the configured history window
``full_backfill`` forces every ticker to re-fetch the full ``full_backfill`` forces every ticker to re-fetch the full
@@ -580,6 +581,7 @@ async def collect_ohlcv(
try: try:
result = await ingestion_service.fetch_and_ingest( result = await ingestion_service.fetch_and_ingest(
db, provider, symbol, start_date=backfill_start, end_date=end_date, db, provider, symbol, start_date=backfill_start, end_date=end_date,
refresh_sr=refresh_sr,
) )
_last_successful[job_name] = symbol _last_successful[job_name] = symbol
processed += 1 processed += 1
@@ -619,6 +621,11 @@ async def collect_ohlcv(
_runtime_finish(job_name, "error", processed=processed, total=total, message=str(exc)) _runtime_finish(job_name, "error", processed=processed, total=total, message=str(exc))
async def collect_ohlcv_for_scan() -> None:
"""Near-close fetch; the scanner immediately rebuilds S/R per ticker."""
await collect_ohlcv(refresh_sr=False)
async def backfill_ohlcv() -> None: async def backfill_ohlcv() -> None:
"""Deep historical backfill: re-fetch the full ``settings.ohlcv_history_days`` """Deep historical backfill: re-fetch the full ``settings.ohlcv_history_days``
window for every ticker, ignoring incremental resume. window for every ticker, ignoring incremental resume.
@@ -1496,8 +1503,8 @@ _DAILY_PIPELINE_STEPS = [
("alerts", "dispatch_alerts_job"), ("alerts", "dispatch_alerts_job"),
] ]
# Near-close (~15:30 ET MonFri): refresh in-progress day-t bars (already how # Near-close (~15:30 ET MonFri): refresh in-progress day-t bars (incremental
# the intraday pipeline keeps the dashboard live), then the only daily # ingestion overlaps the latest stored session), then the only daily
# qualifying R:R scan, then Telegram immediately so manual fills can still hit # qualifying R:R scan, then Telegram immediately so manual fills can still hit
# MOC cutoffs (~15:50/15:55). Under a 15-minute delayed SIP feed a 15:30 scan # MOC cutoffs (~15:50/15:55). Under a 15-minute delayed SIP feed a 15:30 scan
# may see ~15:15 prices — immaterial for a 12-1 momentum signal. # may see ~15:15 prices — immaterial for a 12-1 momentum signal.
@@ -1508,7 +1515,7 @@ _DAILY_PIPELINE_STEPS = [
_NEAR_CLOSE_PIPELINE_STEPS = [ _NEAR_CLOSE_PIPELINE_STEPS = [
# Must land today's in-progress bar (~20 min behind live), or the scan falls # Must land today's in-progress bar (~20 min behind live), or the scan falls
# back to the previous close and execution degrades to the stale_close floor. # back to the previous close and execution degrades to the stale_close floor.
("data_collector", "collect_ohlcv"), ("data_collector", "collect_ohlcv_for_scan"),
("rr_scanner", "scan_rr"), ("rr_scanner", "scan_rr"),
# Straight after the scan so shadow entries mark at the same near-close # Straight after the scan so shadow entries mark at the same near-close
# prices the discretionary book is looking at. # prices the discretionary book is looking at.
+3
View File
@@ -91,3 +91,6 @@ class FundamentalResponse(BaseModel):
metrics: list[MetricItem] | None = None metrics: list[MetricItem] | None = None
valuation: Valuation | None = None valuation: Valuation | None = None
reads: FundamentalsReads | None = None reads: FundamentalsReads | None = None
setup_eligible: bool = True
setup_block_code: str | None = None
setup_block_reason: str | None = None
+4
View File
@@ -53,3 +53,7 @@ class PaperTradeResponse(BaseModel):
# when the trailing exit policy is active. # when the trailing exit policy is active.
trailing_stop: float | None = None trailing_stop: float | None = None
trailing_distance_pct: float | None = None trailing_distance_pct: float | None = None
# Trading sessions represented by post-entry OHLCV bars. These are populated
# only while the active exit policy has a max-hold rule.
sessions_held: int | None = None
sessions_remaining: int | None = None
@@ -0,0 +1,164 @@
"""Actionability gate for incomplete SEC fundamentals."""
from __future__ import annotations
import json
from dataclasses import dataclass
from sqlalchemy import exists, func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.data_import_run import DataImportRun
from app.models.fundamental_snapshot import FundamentalSnapshot
from app.models.sec_filing_gap import SecFilingGap
from app.models.ticker import Ticker
from app.services import fundamental_data_refresh_service
_SEC_FORMS = ("10-K", "10-Q", "10-K/A", "10-Q/A")
@dataclass(frozen=True)
class SetupQuality:
eligible: bool
code: str | None = None
message: str | None = None
async def active_gaps(
db: AsyncSession,
ciks: set[str] | None = None,
) -> list[SecFilingGap]:
"""Unresolved gaps that have not been superseded by a later filing."""
matching_snapshot = exists().where(
FundamentalSnapshot.accession == SecFilingGap.accession
)
gap_date = func.coalesce(
SecFilingGap.index_date,
func.date(SecFilingGap.first_seen_at),
)
later_snapshot = exists().where(
FundamentalSnapshot.cik == SecFilingGap.cik,
FundamentalSnapshot.form.in_(_SEC_FORMS),
FundamentalSnapshot.filed_date > gap_date,
)
stmt = select(SecFilingGap).where(
~matching_snapshot,
~later_snapshot,
)
if ciks is not None:
if not ciks:
return []
stmt = stmt.where(SecFilingGap.cik.in_(ciks))
return list((await db.execute(stmt)).scalars().all())
async def _latest_validation(db: AsyncSession) -> dict:
payload = (
await db.execute(
select(DataImportRun.validation_json)
.where(
DataImportRun.source == "sec_facts",
DataImportRun.validation_json.is_not(None),
)
.order_by(DataImportRun.id.desc())
.limit(1)
)
).scalar_one_or_none()
if not payload:
return {}
try:
summary = json.loads(payload)
except (TypeError, ValueError):
return {}
return summary if isinstance(summary, dict) else {}
async def blocked_reasons_by_cik(
db: AsyncSession,
ciks: set[str] | None = None,
) -> dict[str, str]:
"""Current SEC blocker code by CIK; no historical audit scan."""
if not await fundamental_data_refresh_service.is_enabled(db):
return {}
if ciks is not None and not ciks:
return {}
reasons = {
gap.cik: "sec_filing_gap" for gap in await active_gaps(db, ciks)
}
summary = await _latest_validation(db)
def wanted(cik: str) -> bool:
return ciks is None or cik in ciks
# New summaries carry the complete compact CIK set while the detailed lists
# stay capped for audit readability. Detailed entries supply the reason.
for cik in summary.get("setup_blocked_ciks") or []:
normalized = str(cik) if cik else ""
if normalized and wanted(normalized):
reasons.setdefault(normalized, "sec_filing_gap")
for item in summary.get("missing_xbrl") or []:
normalized = str(item.get("cik") or "")
if normalized and wanted(normalized):
reasons.setdefault(normalized, "sec_filing_gap")
for cik in summary.get("no_xbrl_ciks") or []:
normalized = str(cik) if cik else ""
if normalized and wanted(normalized):
reasons[normalized] = "no_xbrl_filings"
for item in summary.get("no_xbrl_filings") or []:
normalized = str(item.get("cik") or "")
if normalized and wanted(normalized):
reasons[normalized] = "no_xbrl_filings"
return reasons
async def blocked_ciks(db: AsyncSession) -> set[str]:
return set(await blocked_reasons_by_cik(db))
async def blocked_ticker_ids(db: AsyncSession) -> set[int]:
ciks = await blocked_ciks(db)
if not ciks:
return set()
rows = await db.execute(select(Ticker.id).where(Ticker.cik.in_(ciks)))
return {int(ticker_id) for ticker_id in rows.scalars()}
async def ticker_quality(db: AsyncSession, symbol: str) -> SetupQuality:
ticker = (
await db.execute(
select(Ticker).where(Ticker.symbol == symbol.strip().upper())
)
).scalar_one_or_none()
if ticker is None or not ticker.cik:
return SetupQuality(eligible=True)
reason = (await blocked_reasons_by_cik(db, {ticker.cik})).get(ticker.cik)
if reason == "no_xbrl_filings":
return SetupQuality(
eligible=False,
code=reason,
message=(
"No SEC 10-K/10-Q is available for this registrant, so new setups "
"are paused. New registrants clear automatically after their first "
"filing; a successor shell needs an SEC CIK override."
),
)
if reason:
return SetupQuality(
eligible=False,
code=reason,
message=(
"A recent SEC filing is still being reconciled, so new setups are "
"paused. The scheduled fundamentals import retries it automatically."
),
)
return SetupQuality(eligible=True)
async def ticker_is_eligible(db: AsyncSession, ticker_id: int) -> bool:
cik = (
await db.execute(select(Ticker.cik).where(Ticker.id == ticker_id))
).scalar_one_or_none()
if not cik:
return True
return cik not in await blocked_reasons_by_cik(db, {cik})
+29 -3
View File
@@ -100,6 +100,8 @@ async def fetch_and_ingest(
symbol: str, symbol: str,
start_date: date | None = None, start_date: date | None = None,
end_date: date | None = None, end_date: date | None = None,
*,
refresh_sr: bool = True,
) -> IngestionResult: ) -> IngestionResult:
"""Fetch OHLCV data from provider and upsert into Price Store. """Fetch OHLCV data from provider and upsert into Price Store.
@@ -129,7 +131,12 @@ async def fetch_and_ingest(
if bar_count < minimum_backfill_bars: if bar_count < minimum_backfill_bars:
start_date = backfill_start start_date = backfill_start
elif progress is not None: elif progress is not None:
start_date = progress.last_ingested_date + timedelta(days=1) # Re-fetch the latest stored session so an in-progress daily bar can
# be overwritten as the market moves. Starting one day later makes
# every subsequent intraday, near-close, and manual refresh skip
# today's bar once the first partial snapshot has been stored.
# The price-store upsert keeps this one-session overlap idempotent.
start_date = progress.last_ingested_date
else: else:
start_date = backfill_start start_date = backfill_start
@@ -239,7 +246,7 @@ async def fetch_and_ingest(
ticker.symbol, ticker.symbol,
ingested_count, ingested_count,
) )
if ingested_count > 0: if ingested_count > 0 and refresh_sr:
await _refresh_structural_sr(db, ticker.symbol) await _refresh_structural_sr(db, ticker.symbol)
return IngestionResult( return IngestionResult(
symbol=ticker.symbol, symbol=ticker.symbol,
@@ -249,9 +256,28 @@ async def fetch_and_ingest(
message=f"Rate limited. Ingested {ingested_count} records. Resume available.", message=f"Rate limited. Ingested {ingested_count} records. Resume available.",
) )
if ingested_count > 0: if ingested_count > 0 and refresh_sr:
await _refresh_structural_sr(db, ticker.symbol) await _refresh_structural_sr(db, ticker.symbol)
# Incremental fetches deliberately overlap the latest stored session so an
# in-progress bar can be updated. A halted/delisted symbol can therefore
# return one old bar forever; non-empty no longer means fresh. Judge stale
# state from the newest stored session after the upserts instead.
latest = await _get_latest_ohlcv_date(db, ticker.id)
gap_days = (end_date - latest).days if latest is not None else None
if gap_days is not None and gap_days > _STALE_OHLCV_GAP_DAYS:
return IngestionResult(
symbol=ticker.symbol,
records_ingested=ingested_count,
last_date=latest,
status="stale",
message=(
f"No new bars since {latest.isoformat()} ({gap_days}d gap). "
"The symbol may be halted, delisted, or renamed under a new ticker — "
"check the listing and add/fetch the current symbol if it changed."
),
)
return IngestionResult( return IngestionResult(
symbol=ticker.symbol, symbol=ticker.symbol,
records_ingested=ingested_count, records_ingested=ingested_count,
+40 -1
View File
@@ -352,6 +352,7 @@ def _to_dict(
current_price: float | None, current_price: float | None,
benchmark_closes: dict[date, float] | None = None, benchmark_closes: dict[date, float] | None = None,
trailing: tuple[float, float | None] | None = None, trailing: tuple[float, float | None] | None = None,
holding_sessions: tuple[int, int] | None = None,
) -> dict: ) -> dict:
# For open trades, mark to market; for closed, the realized exit price. # For open trades, mark to market; for closed, the realized exit price.
ref = current_price if trade.status == "open" else trade.close_price ref = current_price if trade.status == "open" else trade.close_price
@@ -395,6 +396,8 @@ def _to_dict(
"fill_mode": trade.fill_mode, "fill_mode": trade.fill_mode,
"trailing_stop": trailing[0] if trailing else None, "trailing_stop": trailing[0] if trailing else None,
"trailing_distance_pct": trailing[1] if trailing else None, "trailing_distance_pct": trailing[1] if trailing else None,
"sessions_held": holding_sessions[0] if holding_sessions else None,
"sessions_remaining": holding_sessions[1] if holding_sessions else None,
} }
@@ -435,6 +438,35 @@ async def list_trades(
# Current trailing-stop level + distance for open trades (when a trailing # Current trailing-stop level + distance for open trades (when a trailing
# policy is active). # policy is active).
policy = await get_exit_policy(db) policy = await get_exit_policy(db)
holding_sessions: dict[int, tuple[int, int]] = {}
if policy["mode"] in ("time", "atr_trailing"):
hold_days = int(policy["hold_days"])
open_trades = [trade for trade, _ in rows if trade.status == "open"]
if open_trades:
ticker_ids = {trade.ticker_id for trade in open_trades}
earliest_opened = min(trade.opened_at.date() for trade in open_trades)
session_rows = (
await db.execute(
select(OHLCVRecord.ticker_id, OHLCVRecord.date)
.where(
OHLCVRecord.ticker_id.in_(ticker_ids),
OHLCVRecord.date > earliest_opened,
)
.order_by(OHLCVRecord.ticker_id, OHLCVRecord.date)
)
).all()
dates_by_ticker: dict[int, list[date]] = {}
for ticker_id, session_date in session_rows:
dates_by_ticker.setdefault(int(ticker_id), []).append(session_date)
for trade in open_trades:
dates = dates_by_ticker.get(trade.ticker_id, [])
held = len(dates) - bisect.bisect_right(
dates, trade.opened_at.date()
)
# Do not clamp: a policy shortened below the current holding
# period must remain visible as overdue until the exit pass runs.
holding_sessions[trade.id] = (held, hold_days - held)
trailing_info: dict[int, tuple[float, float | None]] = {} trailing_info: dict[int, tuple[float, float | None]] = {}
if policy["mode"] == "trailing": if policy["mode"] == "trailing":
trail_frac = policy["trailing_pct"] / 100.0 trail_frac = policy["trailing_pct"] / 100.0
@@ -483,7 +515,14 @@ async def list_trades(
trailing_info[t.id] = (level, dist) trailing_info[t.id] = (level, dist)
return [ return [
_to_dict(t, sym, prices.get(t.ticker_id), benchmark_closes, trailing_info.get(t.id)) _to_dict(
t,
sym,
prices.get(t.ticker_id),
benchmark_closes,
trailing_info.get(t.id),
holding_sessions.get(t.id),
)
for t, sym in rows for t, sym in rows
] ]
+63
View File
@@ -27,6 +27,7 @@ from app.models.signal_context_snapshot import SignalContextSnapshot
from app.models.ticker import Ticker from app.models.ticker import Ticker
from app.models.trade_setup import TradeSetup from app.models.trade_setup import TradeSetup
from app.services.indicator_service import _extract_ohlcv, compute_atr from app.services.indicator_service import _extract_ohlcv, compute_atr
from app.services import fundamentals_quality_service, system_event_service
from app.services.price_service import query_ohlcv from app.services.price_service import query_ohlcv
from app.services.qualification import setup_qualifies from app.services.qualification import setup_qualifies
from app.services.sr_service import detect_gate_target_ladder from app.services.sr_service import detect_gate_target_ladder
@@ -526,6 +527,7 @@ async def scan_ticker(
primary_min_rr: float | None = None, primary_min_rr: float | None = None,
gate_levels_override: list[Any] | None = None, gate_levels_override: list[Any] | None = None,
scan_run_id: str | None = None, scan_run_id: str | None = None,
fundamentals_eligible: bool | None = None,
) -> list[TradeSetup]: ) -> list[TradeSetup]:
"""Scan a single ticker for trade setups meeting the R:R threshold. """Scan a single ticker for trade setups meeting the R:R threshold.
@@ -542,6 +544,17 @@ async def scan_ticker(
""" """
ticker = await _get_ticker(db, symbol) ticker = await _get_ticker(db, symbol)
if fundamentals_eligible is None:
fundamentals_eligible = await fundamentals_quality_service.ticker_is_eligible(
db, ticker.id
)
if not fundamentals_eligible:
logger.info(
"Skipping %s: unresolved or unavailable SEC fundamentals",
ticker.symbol,
)
return []
if primary_min_rr is None: if primary_min_rr is None:
primary_min_rr = PRIMARY_TARGET_MIN_RR primary_min_rr = PRIMARY_TARGET_MIN_RR
@@ -726,6 +739,29 @@ async def scan_all_tickers(
ticker_rows = [(int(ticker_id), symbol) for ticker_id, symbol in result.all()] ticker_rows = [(int(ticker_id), symbol) for ticker_id, symbol in result.all()]
total = len(ticker_rows) total = len(ticker_rows)
# Data-quality failures are not weak signals: they make a ticker ineligible.
# Resolve once for the universe scan and pass the decision into scan_ticker.
try:
fundamentals_blocked_ids = (
await fundamentals_quality_service.blocked_ticker_ids(db)
)
except Exception:
await db.rollback()
logger.exception(
"Could not resolve fundamentals quality; blocking this scan closed"
)
await system_event_service.log_event_standalone(
severity="error",
source="rr_scanner",
code="fundamentals_quality_unavailable",
message=(
"The fundamentals quality gate could not be evaluated; the "
"universe scan was blocked to avoid issuing unchecked setups."
),
dedup_key="rr_scanner:fundamentals_quality_unavailable",
)
fundamentals_blocked_ids = {ticker_id for ticker_id, _ in ticker_rows}
# Gate-reset observations must use the same runtime activation settings as # Gate-reset observations must use the same runtime activation settings as
# the live setup list. If the config cannot be loaded, scan normally but do # the live setup list. If the config cannot be loaded, scan normally but do
# not mutate reset state from an evaluation whose rules are unknown. # not mutate reset state from an evaluation whose rules are unknown.
@@ -765,6 +801,12 @@ async def scan_all_tickers(
for index, (ticker_id, symbol) in enumerate(ticker_rows): for index, (ticker_id, symbol) in enumerate(ticker_rows):
if progress_callback is not None: if progress_callback is not None:
progress_callback(index, total, symbol) progress_callback(index, total, symbol)
if ticker_id in fundamentals_blocked_ids:
logger.info(
"Skipping %s: unresolved or unavailable SEC fundamentals",
symbol,
)
continue
# Refresh Structural S/R once, then scores. get_sr_levels is read-only; # Refresh Structural S/R once, then scores. get_sr_levels is read-only;
# without this recalculate the score path would see yesterday's zones. # without this recalculate the score path would see yesterday's zones.
# A refresh failure still scans the ticker: qualification re-gates on # A refresh failure still scans the ticker: qualification re-gates on
@@ -795,6 +837,7 @@ async def scan_all_tickers(
volatility_percentile=(ranks.get(symbol) or {}).get("volatility_percentile"), volatility_percentile=(ranks.get(symbol) or {}).get("volatility_percentile"),
primary_min_rr=PRIMARY_TARGET_MIN_RR, primary_min_rr=PRIMARY_TARGET_MIN_RR,
scan_run_id=scan_run_id, scan_run_id=scan_run_id,
fundamentals_eligible=True,
) )
all_setups.extend(setups) all_setups.extend(setups)
if activation is not None: if activation is not None:
@@ -882,6 +925,26 @@ async def get_trade_setups(
stmt = stmt.where(TradeSetup.recommended_action == recommended_action) stmt = stmt.where(TradeSetup.recommended_action == recommended_action)
excluded_ticker_ids: set[int] = set() excluded_ticker_ids: set[int] = set()
reentry_gate_locks: dict[int, datetime] = {} reentry_gate_locks: dict[int, datetime] = {}
try:
excluded_ticker_ids.update(
await fundamentals_quality_service.blocked_ticker_ids(db)
)
except Exception:
await db.rollback()
logger.exception(
"Could not resolve fundamentals quality; hiding actionable setups"
)
await system_event_service.log_event_standalone(
severity="error",
source="rr_scanner",
code="fundamentals_quality_unavailable",
message=(
"The fundamentals quality gate could not be evaluated; actionable "
"setups were hidden until the metadata check recovers."
),
dedup_key="rr_scanner:fundamentals_quality_unavailable",
)
return []
if exclude_open_trade_tickers: if exclude_open_trade_tickers:
# Manual book only. The shadow book holds the *top-ranked* names by # Manual book only. The shadow book holds the *top-ranked* names by
# construction, so letting its positions hide setups would leave the # construction, so letting its positions hide setups would leave the
+267 -55
View File
@@ -31,10 +31,9 @@ Guardrails (design + reviews):
stored as the parent's. Confirmed 2026-07-27 (NEE via FPL, DOW via Dow Chemical) stored as the parent's. Confirmed 2026-07-27 (NEE via FPL, DOW via Dow Chemical)
and it is not transient: an NEE filing misattributed in 2014 is still misfiled. and it is not transient: an NEE filing misattributed in 2014 is still misfiled.
- **Bounded blocking.** Anything still unresolvable after ``MISSING_XBRL_RETRY_DAYS`` - **Bounded blocking.** Anything still unresolvable after ``MISSING_XBRL_RETRY_DAYS``
stops failing the run and is promoted around, with a named ``unresolved_filing`` stops failing the whole import and enters a durable retry queue. The scheduled
warning. One filing SEC misfiled must not wedge every later import; the index importer retries queued accessions automatically, while the affected issuer is
only moves forward, so a tolerated accession returns only via a reparse, and excluded from actionable setups until its filing is recovered.
only once SEC has re-filed it under the filer's own CIK.
- ``promote`` inserts snapshots ``ON CONFLICT (accession) DO NOTHING`` (immutable), - ``promote`` inserts snapshots ``ON CONFLICT (accession) DO NOTHING`` (immutable),
reports differing existing accessions, and applies ticker updates in the same reports differing existing accessions, and applies ticker updates in the same
transaction. transaction.
@@ -48,18 +47,21 @@ Guardrails (design + reviews):
from __future__ import annotations from __future__ import annotations
import json
import logging import logging
from collections import Counter, defaultdict from collections import Counter, defaultdict
from dataclasses import dataclass, field, replace from dataclasses import dataclass, field, replace
from datetime import date, datetime, timedelta, timezone from datetime import date, datetime, timedelta, timezone
from typing import Any, Callable from typing import Any, Callable
from sqlalchemy import select, update from sqlalchemy import delete, select, update
from app.database import insert_for_session from app.database import insert_for_session
from app.models.data_import_run import DataImportRun from app.models.data_import_run import DataImportRun
from app.models.fundamental_snapshot import FundamentalSnapshot from app.models.fundamental_snapshot import FundamentalSnapshot
from app.models.sec_filing_gap import SecFilingGap
from app.models.system_event import SystemEvent from app.models.system_event import SystemEvent
from app.services import fundamentals_quality_service
from app.services import sec_facts_parser as parser from app.services import sec_facts_parser as parser
from app.services import sec_universe from app.services import sec_universe
from app.services.data_import import STATUS_PROMOTED, ValidationResult from app.services.data_import import STATUS_PROMOTED, ValidationResult
@@ -79,6 +81,7 @@ MIN_BACKFILL_COVERAGE = 0.5
# three); past that it is misfiled, not late, and blocking forever costs more # three); past that it is misfiled, not late, and blocking forever costs more
# than the missing filing does — see the unresolved-filing guardrail below. # than the missing filing does — see the unresolved-filing guardrail below.
MISSING_XBRL_RETRY_DAYS = 3 MISSING_XBRL_RETRY_DAYS = 3
FILING_GAP_ESCALATE_DAYS = 14
# Share-count band a co-registrant-recovered row must land in, relative to the # Share-count band a co-registrant-recovered row must land in, relative to the
# issuer's own last snapshot. Wide enough for buybacks/issuance, nowhere near # issuer's own last snapshot. Wide enough for buybacks/issuance, nowhere near
# wide enough to let a subsidiary shell's token float through (see _shares_continuous). # wide enough to let a subsidiary shell's token float through (see _shares_continuous).
@@ -151,6 +154,7 @@ class SecFundamentalsImporter:
# accession -> the OTHER CIKs the daily index lists it under (co-registrants # accession -> the OTHER CIKs the daily index lists it under (co-registrants
# of a combined filing). Only populated for accessions a tracked issuer filed. # of a combined filing). Only populated for accessions a tracked issuer filed.
self._coregistrants: dict[str, list[int]] = {} self._coregistrants: dict[str, list[int]] = {}
self._retry_rows: list[dict[str, Any]] = []
self._latest_index_date: date | None = None self._latest_index_date: date | None = None
self._backfill = False self._backfill = False
@@ -176,9 +180,19 @@ class SecFundamentalsImporter:
client, last_processed, self._latest_index_date client, last_processed, self._latest_index_date
) )
content = sec_universe.index_content_hash(self._index_rows) content = sec_universe.index_content_hash(self._index_rows)
return sec_universe.compose_revision( revision = sec_universe.compose_revision(
self._latest_index_date, content, self._resolved.symbol_to_cik self._latest_index_date, content, self._resolved.symbol_to_cik
) )
self._retry_rows = []
if not self._backfill:
self._retry_rows = await self._retry_backlog(
db,
set(self._resolved.cik_to_ticker_ids),
)
# Company Facts can change while the daily index revision stays fixed.
# Returning None deliberately bypasses the framework's no-op gate so a
# scheduled run retries every active gap.
return None if self._retry_rows else revision
async def stage(self, db) -> StagedFundamentals: async def stage(self, db) -> StagedFundamentals:
assert self._resolved is not None, "detect_revision must run first" assert self._resolved is not None, "detect_revision must run first"
@@ -193,6 +207,26 @@ class SecFundamentalsImporter:
if r["cik"] in cik_to_tids: if r["cik"] in cik_to_tids:
filed_by_cik[r["cik"]].append(r) filed_by_cik[r["cik"]].append(r)
# Promoted-around filings live in a small durable retry queue, including
# the one-time migration backfill. Merge them into normal incremental
# work so the scheduled importer heals them without operator action.
if not self._backfill:
seen = {
(int(cik), row["accession"])
for cik, rows in filed_by_cik.items()
for row in rows
}
for row in self._retry_rows:
cik = int(row["cik"])
key = (cik, row["accession"])
if key in seen:
continue
filed_by_cik[cik].append(row)
seen.add(key)
coregistrants = [int(value) for value in row.get("coregistrants") or []]
if coregistrants:
self._coregistrants[row["accession"]] = coregistrants
existing = await self._ciks_with_snapshots(db, set(cik_to_tids)) existing = await self._ciks_with_snapshots(db, set(cik_to_tids))
if self._backfill: if self._backfill:
backfill_ciks = set(cik_to_tids) backfill_ciks = set(cik_to_tids)
@@ -247,6 +281,9 @@ class SecFundamentalsImporter:
fiscal_year_end = sub.get("fiscal_year_end") fiscal_year_end = sub.get("fiscal_year_end")
recovered_rows: list[SnapshotRow] = [] recovered_rows: list[SnapshotRow] = []
index_rows = {
row["accession"]: row for row in filed_by_cik.get(cik, [])
}
if is_backfill: if is_backfill:
accns = set(xbrl_meta) accns = set(xbrl_meta)
else: else:
@@ -262,7 +299,13 @@ class SecFundamentalsImporter:
# or a co-registrant filing). NOT a Company-Facts lag — separate # or a co-registrant filing). NOT a Company-Facts lag — separate
# cause, separate fix, so it gets its own reason. # cause, separate fix, so it gets its own reason.
staged.missing_xbrl.append( staged.missing_xbrl.append(
_missing(cik, index_row, "not_in_submissions", self.today) _missing(
cik,
index_row,
"not_in_submissions",
self.today,
self._coregistrants.get(accn),
)
) )
elif accn in present: elif accn in present:
accns.add(accn) accns.add(accn)
@@ -289,11 +332,26 @@ class SecFundamentalsImporter:
"coregistrant_facts_rejected" if source_cik "coregistrant_facts_rejected" if source_cik
else "not_in_companyfacts", else "not_in_companyfacts",
self.today, self.today,
self._coregistrants.get(accn),
)) ))
# fiscalYearEnd (MMDD) is what lets the parser derive period identity from # fiscalYearEnd (MMDD) is what lets the parser derive period identity from
# reportDate instead of SEC's unreliable fy/fp fields. # reportDate instead of SEC's unreliable fy/fp fields.
result = parser.parse_snapshots(cf, xbrl_meta, accns, fiscal_year_end=fiscal_year_end) result = parser.parse_snapshots(cf, xbrl_meta, accns, fiscal_year_end=fiscal_year_end)
for skipped in result.skipped_filings:
index_row = index_rows.get(skipped["accession"])
if index_row is not None:
# Facts are present but our parser cannot construct a snapshot.
# A new index row keeps the normal grace period before promotion;
# a row already read from the queue retains its _retry_queue marker
# so later imports promote and retry without wedging the index.
staged.missing_xbrl.append(_missing(
cik,
index_row,
"parser_unusable",
self.today,
self._coregistrants.get(skipped["accession"]),
))
staged.rows.extend(result.rows) staged.rows.extend(result.rows)
staged.rows.extend(recovered_rows) staged.rows.extend(recovered_rows)
staged.skipped_filings.extend(result.skipped_filings) staged.skipped_filings.extend(result.skipped_filings)
@@ -398,11 +456,22 @@ class SecFundamentalsImporter:
"skipped_non_xbrl": len(staged.skipped_non_xbrl), "skipped_non_xbrl": len(staged.skipped_non_xbrl),
"no_xbrl_filings": staged.no_xbrl_filings[:50], "no_xbrl_filings": staged.no_xbrl_filings[:50],
"no_xbrl_filings_count": len(staged.no_xbrl_filings), "no_xbrl_filings_count": len(staged.no_xbrl_filings),
"no_xbrl_ciks": sorted({
str(item["cik"])
for item in staged.no_xbrl_filings
if item.get("cik")
}),
"missing_xbrl": staged.missing_xbrl[:50], "missing_xbrl": staged.missing_xbrl[:50],
"missing_xbrl_count": len(staged.missing_xbrl), "missing_xbrl_count": len(staged.missing_xbrl),
"missing_xbrl_blocking": len(blocking), "missing_xbrl_blocking": len(blocking),
"recovered_from_coregistrant": staged.recovered[:50], "recovered_from_coregistrant": staged.recovered[:50],
"recovered_count": len(staged.recovered), "recovered_count": len(staged.recovered),
# Complete compact gate input; detailed audit lists above stay capped.
"setup_blocked_ciks": sorted({
str(item["cik"])
for item in [*staged.missing_xbrl, *staged.no_xbrl_filings]
if item.get("cik")
}),
"invalid_payloads": staged.invalid_payloads, "invalid_payloads": staged.invalid_payloads,
"cik_updates": len(staged.resolved.cik_updates), "cik_updates": len(staged.resolved.cik_updates),
# differing existing accessions (immutable — kept, reported here) # differing existing accessions (immutable — kept, reported here)
@@ -420,14 +489,17 @@ class SecFundamentalsImporter:
retryable=( retryable=(
len(messages) == 1 len(messages) == 1
and bool(blocking) and bool(blocking)
and all(m.get("reason") == "not_in_companyfacts" for m in blocking) and all(
m.get("reason") in {"not_in_companyfacts", "parser_unusable"}
for m in blocking
)
), ),
deferred_alert_after_days=MISSING_XBRL_RETRY_DAYS, deferred_alert_after_days=MISSING_XBRL_RETRY_DAYS,
deferred_alert_messages=( deferred_alert_messages=(
[ [
f"{len(aged_out)} tracked SEC filing(s) remain unresolved past " f"{len(aged_out)} tracked SEC filing(s) remain unresolved past "
f"the {MISSING_XBRL_RETRY_DAYS}-day retry window and risk being " f"the {MISSING_XBRL_RETRY_DAYS}-day retry window. They will "
f"promoted around without automatic retry: " f"enter automatic retry and block affected symbols from setups: "
f"{_missing_detail(aged_out)}" f"{_missing_detail(aged_out)}"
] ]
if aged_out if aged_out
@@ -462,6 +534,69 @@ class SecFundamentalsImporter:
await db.execute(stmt) await db.execute(stmt)
inserted += 1 inserted += 1
# Synchronize the retry queue in the snapshot-promotion transaction.
existing_gaps = (await db.execute(select(SecFilingGap))).scalars().all()
existing_gap_accessions = {gap.accession for gap in existing_gaps}
resolved_accessions = {row.accession for row in staged.rows}
# A filing now classified non-XBRL can never yield a snapshot and is no
# longer a fundamentals completeness gap.
resolved_accessions.update(
item["accession"] for item in staged.skipped_non_xbrl
)
queue_resolved = 0
if resolved_accessions:
result = await db.execute(
delete(SecFilingGap).where(
SecFilingGap.accession.in_(resolved_accessions)
)
)
queue_resolved = int(result.rowcount or 0)
now = _now()
tolerated = _past_retry_window(staged.missing_xbrl)
for gap in tolerated:
stmt = insert_for_session(db, SecFilingGap).values(
cik=gap["cik"],
accession=gap["accession"],
form=gap.get("form"),
index_date=gap.get("index_date"),
reason=gap["reason"],
coregistrant_ciks_json=json.dumps(gap.get("coregistrants") or []),
first_seen_at=now,
last_attempted_at=now,
)
await db.execute(
stmt.on_conflict_do_update(
index_elements=["accession"],
set_={
"cik": stmt.excluded.cik,
"form": stmt.excluded.form,
"index_date": stmt.excluded.index_date,
"reason": stmt.excluded.reason,
"coregistrant_ciks_json": stmt.excluded.coregistrant_ciks_json,
"last_attempted_at": stmt.excluded.last_attempted_at,
},
)
)
# Remove gaps made irrelevant by a later valid 10-K/10-Q. Quality reads
# already ignore them; physical cleanup keeps the queue small.
active_ids = {gap.id for gap in await fundamentals_quality_service.active_gaps(db)}
obsolete_ids = {
gap.id for gap in existing_gaps
if gap.id not in active_ids and gap.accession not in resolved_accessions
}
if obsolete_ids:
result = await db.execute(
delete(SecFilingGap).where(SecFilingGap.id.in_(obsolete_ids))
)
queue_resolved += int(result.rowcount or 0)
newly_queued = [
gap for gap in tolerated
if gap["accession"] not in existing_gap_accessions
]
# Warn (in-transaction, so it commits atomically with the promotion) when # Warn (in-transaction, so it commits atomically with the promotion) when
# any existing accession reconstructed differently — kept immutable. # any existing accession reconstructed differently — kept immutable.
if staged.discrepancies: if staged.discrepancies:
@@ -481,69 +616,87 @@ class SecFundamentalsImporter:
created_at=_now(), created_at=_now(),
)) ))
# Persistent current gaps get one actionable escalation rather than a
# daily warning. The nullable marker makes this durable and noise-free.
escalation_cutoff = now - timedelta(days=FILING_GAP_ESCALATE_DAYS)
aged_gaps = (
await db.execute(
select(SecFilingGap).where(
SecFilingGap.first_seen_at <= escalation_cutoff,
SecFilingGap.escalated_at.is_(None),
)
)
).scalars().all()
if aged_gaps:
named = ", ".join(
f"{gap.cik}/{gap.accession} ({gap.reason})"
for gap in aged_gaps[:10]
)
db.add(SystemEvent(
severity="warning",
source="sec_facts",
code="filing_gap_aged",
message=(
f"{len(aged_gaps)} SEC filing gap(s) remain unresolved after "
f"{FILING_GAP_ESCALATE_DAYS} days; affected setups remain paused. "
f"Review the filing/CIK mapping or parser: {named}"
)[:4000],
dedup_key=f"sec_facts:filing_gap_aged:{run_id}",
created_at=now,
))
await db.execute(
update(SecFilingGap)
.where(SecFilingGap.id.in_([gap.id for gap in aged_gaps]))
.values(escalated_at=now)
)
# Recovered rows are real data from an unexpected place — record where they # Recovered rows are real data from an unexpected place — record where they
# came from, so a wrong recovery is auditable rather than invisible. # came from, so a wrong recovery is auditable rather than invisible.
if staged.recovered: if staged.recovered:
named = ", ".join( named = ", ".join(
f"{r['accession']} <- CIK {r['source_cik']}" for r in staged.recovered[:10] f"{r['accession']} <- CIK {r['source_cik']}" for r in staged.recovered[:10]
) )
db.add(SystemEvent( logger.info(
severity="warning", "sec_facts: recovered %d filing(s) from co-registrants: %s",
source="sec_facts", len(staged.recovered),
code="coregistrant_recovery", named,
message=( )
f"{len(staged.recovered)} filing(s) were absent from the filer's "
f"own Company Facts and were parsed from a co-registrant's file "
f"instead (share count checked against the issuer's history): {named}"
)[:4000],
dedup_key=f"sec_facts:coregistrant_recovery:{run_id}",
created_at=_now(),
))
# Filings past the retry window: promoted WITHOUT them so one misfiled # One warning when a gap first enters automatic retry. Repeating it every
# filing cannot wedge every later import. This is the deliberate trade — # day adds noise; the queue remains the durable actionable state.
# loud and named, because the index only moves forward and nothing will if newly_queued:
# revisit them on its own. symbols_by_cik: dict[str, list[str]] = defaultdict(list)
tolerated = _past_retry_window(staged.missing_xbrl) for symbol, cik in staged.resolved.symbol_to_cik.items():
if tolerated: symbols_by_cik[cik10(cik)].append(symbol)
named = ", ".join(
f"{'/'.join(symbols_by_cik.get(gap['cik'], [])) or gap['cik']}"
f"/{gap['accession']}"
for gap in newly_queued[:10]
)
db.add(SystemEvent( db.add(SystemEvent(
severity="warning", severity="warning",
source="sec_facts", source="sec_facts",
code="unresolved_filing", code="unresolved_filing",
message=( message=(
f"{len(tolerated)} tracked filing(s) still unresolvable after " f"{len(newly_queued)} filing(s) entered automatic SEC retry. "
f"{MISSING_XBRL_RETRY_DAYS} days; promoting without them rather " f"Affected symbols are blocked from new actionable setups until "
f"than blocking every later import. They are NOT retried. " f"their filing is recovered: {named}"
f"scripts/reparse_fundamentals.py recovers them ONLY once SEC "
f"re-files under the filer's own CIK — it walks no index, so it "
f"cannot reach facts still sitting under a co-registrant: "
f"{_missing_detail(tolerated)}"
)[:4000], )[:4000],
dedup_key=f"sec_facts:unresolved_filing:{run_id}", dedup_key=f"sec_facts:unresolved_filing:{run_id}",
created_at=_now(), created_at=_now(),
)) ))
# A tracked issuer whose registrant has no XBRL filings can never produce a # A new registrant may have no XBRL filing yet. Keep it out of actionable
# snapshot, and it is restaged on every run forever. That is a resolution # setups, but log it instead of raising a recurring operator warning.
# problem, not missing data, and it is silent without this.
if staged.no_xbrl_filings: if staged.no_xbrl_filings:
named = ", ".join( named = ", ".join(
f"{e['cik']} ({e.get('name') or '?'})" for e in staged.no_xbrl_filings[:10] f"{e['cik']} ({e.get('name') or '?'})" for e in staged.no_xbrl_filings[:10]
) )
db.add(SystemEvent( logger.info(
severity="warning", "sec_facts: %d registrant(s) have no XBRL history yet: %s",
source="sec_facts", len(staged.no_xbrl_filings),
code="no_xbrl_filings", named,
message=( )
f"{len(staged.no_xbrl_filings)} tracked issuer(s) resolved to a "
f"registrant with no XBRL 10-K/10-Q. Either a successor shell "
f"(pin the real filer via the '{sec_universe.CIK_OVERRIDES_KEY}' "
f"setting) or a new registrant that has not filed its first "
f"10-K/10-Q yet, which needs nothing and clears itself: {named}"
)[:4000],
dedup_key=f"sec_facts:no_xbrl_filings:{run_id}",
created_at=_now(),
))
ticker_counts = await sec_universe.apply_ticker_updates( ticker_counts = await sec_universe.apply_ticker_updates(
db, staged.resolved, staged.sic_updates db, staged.resolved, staged.sic_updates
@@ -553,11 +706,57 @@ class SecFundamentalsImporter:
"updated": updated, "updated": updated,
"existing_unchanged": len(staged.existing_accessions) - updated, "existing_unchanged": len(staged.existing_accessions) - updated,
"discrepancies": len(staged.discrepancies), "discrepancies": len(staged.discrepancies),
"retry_queue_added": len(newly_queued),
"retry_queue_resolved": queue_resolved,
**ticker_counts, **ticker_counts,
} }
# -- helpers ----------------------------------------------------------- # -- helpers -----------------------------------------------------------
async def _retry_backlog(
self,
db,
tracked_ciks: set[int],
) -> list[dict[str, Any]]:
"""Active typed gaps; migration 028 owns historical bootstrap."""
if not tracked_ciks:
return []
tracked = {cik10(cik) for cik in tracked_ciks}
candidates: dict[str, dict[str, Any]] = {}
queued = await fundamentals_quality_service.active_gaps(db, tracked)
for gap in queued:
try:
coregistrants = json.loads(gap.coregistrant_ciks_json or "[]")
except (TypeError, ValueError):
coregistrants = []
candidates[gap.accession] = {
"cik": gap.cik,
"accession": gap.accession,
"form": gap.form,
"index_date": gap.index_date,
"reason": gap.reason,
"coregistrants": coregistrants,
"_retry_queue": True,
}
if not candidates:
return []
resolved = set(
(
await db.execute(
select(FundamentalSnapshot.accession).where(
FundamentalSnapshot.accession.in_(list(candidates))
)
)
).scalars().all()
)
return [
item
for accession, item in candidates.items()
if accession not in resolved
]
async def _last_processed_index_date(self, db) -> date | None: async def _last_processed_index_date(self, db) -> date | None:
return ( return (
await db.execute( await db.execute(
@@ -666,19 +865,32 @@ def _companyfacts_structure_error(cf: Any) -> str | None:
return None return None
def _missing(cik: int, row: dict[str, Any], reason: str, today: date) -> dict[str, Any]: def _missing(
cik: int,
row: dict[str, Any],
reason: str,
today: date,
coregistrants: list[int] | None = None,
) -> dict[str, Any]:
"""One unresolvable index row, carrying everything needed to look the filing """One unresolvable index row, carrying everything needed to look the filing
up by hand (EDGAR accession + the index date it was seen on) and to decide up by hand (EDGAR accession + the index date it was seen on) and to decide
whether it is still young enough to be worth blocking on.""" whether it is still young enough to be worth blocking on."""
index_date = row.get("index_date") index_date = row.get("index_date")
age_days = (
(today - index_date).days if isinstance(index_date, date) else 0
)
if row.get("_retry_queue"):
age_days = max(age_days, MISSING_XBRL_RETRY_DAYS + 1)
return { return {
"cik": cik10(cik), "cik": cik10(cik),
"accession": row["accession"], "accession": row["accession"],
"form": row.get("form"), "form": row.get("form"),
"index_date": index_date, "index_date": index_date,
# No index date (older cached rows) => age 0 => blocks, the safe default. # A newly observed row without a date blocks safely. A durable queue row
"age_days": (today - index_date).days if isinstance(index_date, date) else 0, # has already passed the bounded window and is forced aged-out above.
"age_days": age_days,
"reason": reason, "reason": reason,
"coregistrants": list(coregistrants or []),
} }
+15 -1
View File
@@ -15,6 +15,11 @@ not add OS cron entries: the application scheduler owns both jobs.
- Cron expressions are editable in Admin → Schedule. - Cron expressions are editable in Admin → Schedule.
- Every attempt is recorded in `data_import_runs`; failures also create a system - Every attempt is recorded in `data_import_runs`; failures also create a system
event. A failed validation does not promote partial data. event. A failed validation does not promote partial data.
- An SEC filing still missing after the short publication-lag window enters
`sec_filing_gaps`. The daily importer retries it automatically; affected
tickers are excluded from actionable setups until a snapshot is recovered or
a later valid 10-K/10-Q supersedes the gap. Migration `028` materializes older
promoted gaps into this queue once, so setup reads never scan import history.
The systemd service uses one application worker. The import framework also holds The systemd service uses one application worker. The import framework also holds
a PostgreSQL advisory lock per source, so an overlapping manual/scheduled run is a PostgreSQL advisory lock per source, so an overlapping manual/scheduled run is
@@ -83,7 +88,8 @@ In Admin → Jobs, wait until no other job is running, then:
3. Check Admin → System Events. There should be no new import error. 3. Check Admin → System Events. There should be no new import error.
4. Confirm the next-run times correspond to 02:30 and 04:00 New York time. 4. Confirm the next-run times correspond to 02:30 and 04:00 New York time.
5. Open several ticker pages and confirm the fundamentals panel has populated 5. Open several ticker pages and confirm the fundamentals panel has populated
data and still handles partial/missing issuers cleanly. data and still handles partial/missing issuers cleanly. A ticker held by the
quality gate should show **New setups paused** with the specific SEC reason.
## A5 parity observation window ## A5 parity observation window
@@ -231,6 +237,14 @@ least several scheduled cycles before A6 removes the legacy providers.
must stop. Existing promoted snapshots/events remain available. must stop. Existing promoted snapshots/events remain available.
- Inspect the job runtime, latest `data_import_runs.validation_json`, service - Inspect the job runtime, latest `data_import_runs.validation_json`, service
logs, and Admin → System Events before retrying. logs, and Admin → System Events before retrying.
- `unresolved_filing` is emitted once when a filing enters automatic retry. It
does not require a server command. If the gap is still current after 14 days,
`filing_gap_aged` is emitted once with the CIK, accession, and parser/mapping
reason. A later valid 10-K/10-Q retires the gap even when the original SEC
accession never becomes usable.
- Successful co-registrant recovery is logged without a warning. New registrants
with no XBRL history are also logged quietly, but their ticker page explains
that setups remain paused and that successor shells may need `sec_cik_overrides`.
- Re-run `sudo -u deploy bash ./deploy/provision_fundamentals.sh --check` for - Re-run `sudo -u deploy bash ./deploy/provision_fundamentals.sh --check` for
binary, clone, permission, disk, or environment failures. binary, clone, permission, disk, or environment failures.
- The Dolt clone is a reproducible cache and does not need a bespoke backup. - The Dolt clone is a reproducible cache and does not need a bespoke backup.
@@ -22,6 +22,24 @@ function pnlColor(v: number): string {
return 'text-gray-300'; return 'text-gray-300';
} }
function maxHoldText(trade: PaperTrade, compact = false): string | null {
const remaining = trade.sessions_remaining;
if (remaining == null) return null;
const held = trade.sessions_held ?? 0;
if (remaining < 0) return compact ? 'past max hold' : `${held} held · past max hold`;
if (remaining === 0) return compact ? 'max hold reached' : `${held} held · max hold reached`;
if (compact) return `${remaining} ${remaining === 1 ? 'session' : 'sessions'} left`;
return `${held} held · ${remaining} remaining`;
}
function maxHoldColor(trade: PaperTrade): string {
const remaining = trade.sessions_remaining;
if (remaining == null) return 'text-gray-400';
const holdDays = Math.max(1, (trade.sessions_held ?? 0) + remaining);
const warningAt = Math.max(1, Math.ceil(holdDays * 0.2));
return remaining <= warningAt ? 'text-amber-300' : 'text-gray-400';
}
function DirTag({ direction }: { direction: string }) { function DirTag({ direction }: { direction: string }) {
const isLong = direction === 'long'; const isLong = direction === 'long';
return ( return (
@@ -116,6 +134,13 @@ function TradeDetail({ trade, exitLabel, exitMode, atrMultiplier, trailingPct, o
} }
/> />
<Detail label="exit rule" value={exitLabel ?? 'target/stop'} /> <Detail label="exit rule" value={exitLabel ?? 'target/stop'} />
{maxHoldText(trade) && (
<Detail
label="max hold"
value={maxHoldText(trade)}
valueClass={maxHoldColor(trade)}
/>
)}
<div className="flex items-end"> <div className="flex items-end">
<button <button
onClick={onClose} onClick={onClose}
@@ -173,13 +198,14 @@ export function OpenTradesPanel() {
const trailingPct = policy?.trailing_pct ?? 12; const trailingPct = policy?.trailing_pct ?? 12;
const exitLabel = policy const exitLabel = policy
? policy.mode === 'atr_trailing' ? policy.mode === 'atr_trailing'
? `${atrMultiplier.toFixed(1)}x ATR trailing stop / ${policy.hold_days}d max` ? `${atrMultiplier.toFixed(1)}x ATR trailing stop / ${policy.hold_days} sessions max`
: policy.mode === 'trailing' : policy.mode === 'trailing'
? `trailing ${Math.round(trailingPct)}%` ? `trailing ${Math.round(trailingPct)}%`
: policy.mode === 'time' : policy.mode === 'time'
? `${policy.hold_days}d hold` ? `${policy.hold_days}-session hold`
: 'target/stop' : 'target/stop'
: null; : null;
const hasMaxHold = exitMode === 'atr_trailing' || exitMode === 'time';
const rows = trades ?? []; const rows = trades ?? [];
@@ -217,6 +243,7 @@ export function OpenTradesPanel() {
{rows.map((t) => { {rows.map((t) => {
const p = tradePnl(t); const p = tradePnl(t);
const open = expandedId === t.id; const open = expandedId === t.id;
const holdText = maxHoldText(t, true);
return ( return (
<li key={t.id}> <li key={t.id}>
<div <div
@@ -230,7 +257,11 @@ export function OpenTradesPanel() {
} }
}} }}
aria-expanded={open} aria-expanded={open}
className="grid w-full cursor-pointer grid-cols-[110px_1fr_60px_16px] items-center gap-3 rounded-lg px-2 py-2.5 text-left transition-colors hover:bg-white/[0.03] sm:grid-cols-[130px_150px_1fr_70px_16px]" className={`grid w-full cursor-pointer grid-cols-[110px_1fr_60px_16px] items-center gap-3 rounded-lg px-2 py-2.5 text-left transition-colors hover:bg-white/[0.03] ${
hasMaxHold
? 'sm:grid-cols-[130px_150px_1fr_70px_16px] lg:grid-cols-[130px_150px_1fr_110px_70px_16px]'
: 'sm:grid-cols-[130px_150px_1fr_70px_16px]'
}`}
> >
<span className="flex items-center gap-2"> <span className="flex items-center gap-2">
<Link <Link
@@ -246,6 +277,14 @@ export function OpenTradesPanel() {
{formatPrice(t.entry_price)} {t.current_price != null ? formatPrice(t.current_price) : '—'} {formatPrice(t.entry_price)} {t.current_price != null ? formatPrice(t.current_price) : '—'}
</span> </span>
<RBar r={p?.r ?? null} max={rMax} /> <RBar r={p?.r ?? null} max={rMax} />
{hasMaxHold && (
<span
className={`num hidden text-right text-[11px] lg:block ${maxHoldColor(t)}`}
title="Maximum holding period; the stop may close this trade sooner."
>
{holdText ?? '—'}
</span>
)}
<span className={`num text-right text-[13px] font-semibold ${p?.r != null ? pnlColor(p.r) : 'text-gray-500'}`}> <span className={`num text-right text-[13px] font-semibold ${p?.r != null ? pnlColor(p.r) : 'text-gray-500'}`}>
{p?.r != null ? `${p.r >= 0 ? '+' : ''}${p.r.toFixed(2)}R` : '—'} {p?.r != null ? `${p.r >= 0 ? '+' : ''}${p.r.toFixed(2)}R` : '—'}
</span> </span>
+1
View File
@@ -38,6 +38,7 @@ const ind = (median: number, favorable_percentile: number) =>
const legacy = { const legacy = {
pe_ratio: null, revenue_growth: null, earnings_surprise: null, market_cap: null, pe_ratio: null, revenue_growth: null, earnings_surprise: null, market_cap: null,
next_earnings_date: null, fetched_at: null, unavailable_fields: {}, next_earnings_date: null, fetched_at: null, unavailable_fields: {},
setup_eligible: true, setup_block_code: null, setup_block_reason: null,
}; };
const full: FundamentalResponse = { const full: FundamentalResponse = {
+5
View File
@@ -237,6 +237,8 @@ export interface PaperTrade {
close_reason: 'time' | 'trailing' | 'stop' | 'target' | 'manual' | null; close_reason: 'time' | 'trailing' | 'stop' | 'target' | 'manual' | null;
trailing_stop: number | null; trailing_stop: number | null;
trailing_distance_pct: number | null; trailing_distance_pct: number | null;
sessions_held: number | null;
sessions_remaining: number | null;
} }
export interface ExitPolicy { export interface ExitPolicy {
@@ -827,6 +829,9 @@ export interface FundamentalResponse {
metrics: MetricItem[] | null; metrics: MetricItem[] | null;
valuation: Valuation | null; valuation: Valuation | null;
reads: FundamentalsReads | null; reads: FundamentalsReads | null;
setup_eligible: boolean;
setup_block_code: string | null;
setup_block_reason: string | null;
} }
// Indicators // Indicators
+52 -3
View File
@@ -64,10 +64,44 @@ function timeAgo(iso: string): string {
return `${days}d ago`; return `${days}d ago`;
} }
function marketDate(date = new Date()): string {
const parts = new Intl.DateTimeFormat('en-US', {
timeZone: 'America/New_York',
year: 'numeric',
month: '2-digit',
day: '2-digit',
}).formatToParts(date);
const value = (type: Intl.DateTimeFormatPartTypes) =>
parts.find((part) => part.type === type)?.value ?? '';
return value('year') + '-' + value('month') + '-' + value('day');
}
function formatSessionDate(isoDate: string): string {
const currentMarketDate = marketDate();
if (isoDate === currentMarketDate) return 'Today';
// Parse date-only market sessions explicitly. Parsing YYYY-MM-DD directly as
// a Date means midnight UTC and makes today's bar look many hours old.
const [year, month, day] = isoDate.split('-').map(Number);
if (!year || !month || !day) return isoDate;
return new Intl.DateTimeFormat(undefined, {
month: 'short',
day: 'numeric',
year: year === Number(currentMarketDate.slice(0, 4)) ? undefined : 'numeric',
timeZone: 'UTC',
}).format(new Date(Date.UTC(year, month - 1, day)));
}
function formatOHLCVFreshness(sessionDate: string, updatedAt?: string | null): string {
const session = formatSessionDate(sessionDate);
return updatedAt ? session + ' · updated ' + timeAgo(updatedAt) : session;
}
interface DataStatusItem { interface DataStatusItem {
label: string; label: string;
available: boolean; available: boolean;
timestamp?: string | null; timestamp?: string | null;
timestampLabel?: string | null;
selector: FetchSelector; // what a refresh of this row fetches selector: FetchSelector; // what a refresh of this row fetches
paid?: boolean; // provider call that may cost money/quota paid?: boolean; // provider call that may cost money/quota
} }
@@ -100,7 +134,7 @@ function DataFreshnessBar({
}`} /> }`} />
<span className="text-xs text-gray-400">{item.label}</span> <span className="text-xs text-gray-400">{item.label}</span>
{item.available && item.timestamp ? ( {item.available && item.timestamp ? (
<span className="text-[10px] text-gray-500">{timeAgo(item.timestamp)}</span> <span className="text-[10px] text-gray-500">{item.timestampLabel ?? timeAgo(item.timestamp)}</span>
) : !item.available ? ( ) : !item.available ? (
<span className="text-[10px] text-gray-600">no data</span> <span className="text-[10px] text-gray-600">no data</span>
) : null} ) : null}
@@ -171,10 +205,16 @@ export default function TickerDetailPage() {
const dataStatus: DataStatusItem[] = useMemo(() => [ const dataStatus: DataStatusItem[] = useMemo(() => [
{ {
label: 'OHLCV', label: 'OHLCV',
// Market age of the latest bar (session date), not DB insert time — // Keep the market session date distinct from the last successful bar
// created_at stays frozen when the provider returns no new sessions. // write; treating YYYY-MM-DD as an instant makes today's session look old.
available: !!ohlcv.data && ohlcv.data.length > 0, available: !!ohlcv.data && ohlcv.data.length > 0,
timestamp: ohlcv.data?.[ohlcv.data.length - 1]?.date, timestamp: ohlcv.data?.[ohlcv.data.length - 1]?.date,
timestampLabel: ohlcv.data?.length
? formatOHLCVFreshness(
ohlcv.data[ohlcv.data.length - 1].date,
ohlcv.data[ohlcv.data.length - 1].created_at,
)
: null,
selector: ['ohlcv'] as FetchSelector, selector: ['ohlcv'] as FetchSelector,
paid: true, paid: true,
}, },
@@ -319,6 +359,15 @@ export default function TickerDetailPage() {
busy={ingestion.isPending} busy={ingestion.isPending}
/> />
</div> </div>
{fundamentals.data && !fundamentals.data.setup_eligible && (
<div className="border-b border-white/[0.06] px-6 py-3 sm:px-7">
<Callout variant="warning">
<span className="font-medium">New setups paused.</span>{' '}
{fundamentals.data.setup_block_reason ??
'SEC fundamentals are incomplete for this ticker.'}
</Callout>
</div>
)}
<div className="p-6 pb-5 sm:p-7 sm:pb-5"> <div className="p-6 pb-5 sm:p-7 sm:pb-5">
<div className="flex flex-wrap items-start justify-between gap-x-8 gap-y-5"> <div className="flex flex-wrap items-start justify-between gap-x-8 gap-y-5">
<div className="min-w-0"> <div className="min-w-0">
+21 -4
View File
@@ -15,6 +15,8 @@ from sqlalchemy.ext.asyncio import (
create_async_engine, create_async_engine,
) )
from sqlalchemy import delete
from app.database import Base from app.database import Base
from app.providers.protocol import OHLCVData from app.providers.protocol import OHLCVData
@@ -32,14 +34,29 @@ _test_session_factory = async_sessionmaker(
) )
_schema_created = False
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
async def _setup_db(): async def _setup_db():
"""Create all tables before each test and drop them after.""" """Hand every test an empty database.
The schema is built once and then truncated per test rather than dropped and
recreated. A create_all/drop_all cycle costs ~49ms against these 22 tables and
ran for every test in the suite — including the many that never open a session
— where deleting every row costs ~6ms for the same guarantee. No model sets
``sqlite_autoincrement``, so SQLite reuses rowids after a full delete and
generated ids still restart at 1.
"""
global _schema_created
async with _test_engine.begin() as conn: async with _test_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all) if not _schema_created:
await conn.run_sync(Base.metadata.create_all)
_schema_created = True
else:
for table in reversed(Base.metadata.sorted_tables):
await conn.execute(delete(table))
yield yield
async with _test_engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
@pytest.fixture @pytest.fixture
@@ -0,0 +1,183 @@
from __future__ import annotations
import json
from datetime import date, datetime, timezone
from app.models.data_import_run import DataImportRun
from app.models.fundamental_snapshot import FundamentalSnapshot
from app.models.sec_filing_gap import SecFilingGap
from app.models.settings import SystemSetting
from app.models.ticker import Ticker
from app.services import fundamentals_quality_service
async def test_latest_sec_validation_blocks_deferred_and_no_history_ciks(
db_session,
):
missing = Ticker(symbol="MISSING", cik="0000000001")
no_history = Ticker(symbol="NEWREG", cik="0000000002")
healthy = Ticker(symbol="HEALTHY", cik="0000000003")
db_session.add_all([missing, no_history, healthy])
await db_session.flush()
db_session.add(
SystemSetting(
key="fundamental_data_sec_dolt_cutover_enabled",
value="true",
)
)
db_session.add(
DataImportRun(
source="sec_facts",
status="deferred",
validation_json=json.dumps({
"missing_xbrl": [{"cik": missing.cik, "accession": "MISSING-Q"}],
"no_xbrl_filings": [{"cik": no_history.cik}],
}),
started_at=datetime.now(timezone.utc),
)
)
await db_session.flush()
assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == {
missing.id,
no_history.id,
}
async def test_sec_quality_gate_is_inactive_before_cutover(db_session):
ticker = Ticker(symbol="SHADOW", cik="0000000042")
db_session.add(ticker)
await db_session.flush()
now = datetime.now(timezone.utc)
db_session.add(
SecFilingGap(
cik=ticker.cik,
accession="SHADOW-Q",
form="10-Q",
index_date=date.today(),
reason="not_in_companyfacts",
first_seen_at=now,
last_attempted_at=now,
)
)
await db_session.flush()
assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == set()
async def test_active_gap_is_blocked_until_a_later_filing_supersedes_it(db_session):
ticker = Ticker(symbol="HIST", cik="0000000043")
now = datetime.now(timezone.utc)
db_session.add_all([
ticker,
SystemSetting(
key="fundamental_data_sec_dolt_cutover_enabled",
value="true",
),
SecFilingGap(
cik=ticker.cik,
accession="HIST-Q",
form="10-Q",
index_date=date.today().replace(day=1),
reason="coregistrant_facts_rejected",
first_seen_at=now,
last_attempted_at=now,
),
])
await db_session.flush()
assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == {
ticker.id
}
db_session.add(
FundamentalSnapshot(
cik=ticker.cik,
accession="LATER-Q",
form="10-Q",
filed_date=date.today(),
accepted_at=datetime.now(timezone.utc),
period_end=date.today(),
fiscal_year=date.today().year,
fiscal_period="Q2",
)
)
await db_session.flush()
assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == set()
async def test_gap_without_index_date_uses_first_seen_date_for_supersession(
db_session,
):
ticker = Ticker(symbol="DATELESS", cik="0000000045")
first_seen = datetime(2026, 5, 1, 12, tzinfo=timezone.utc)
db_session.add_all([
ticker,
SystemSetting(
key="fundamental_data_sec_dolt_cutover_enabled",
value="true",
),
SecFilingGap(
cik=ticker.cik,
accession="DATELESS-Q",
form="10-Q",
index_date=None,
reason="not_in_companyfacts",
first_seen_at=first_seen,
last_attempted_at=first_seen,
),
])
await db_session.flush()
assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == {
ticker.id
}
db_session.add(
FundamentalSnapshot(
cik=ticker.cik,
accession="LATER-DATELESS-Q",
form="10-Q",
filed_date=date(2026, 5, 2),
accepted_at=datetime(2026, 5, 2, 12, tzinfo=timezone.utc),
period_end=date(2026, 3, 31),
fiscal_year=2026,
fiscal_period="Q1",
)
)
await db_session.flush()
assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == set()
async def test_ticker_quality_explains_no_xbrl_block(db_session):
ticker = Ticker(symbol="NEWREG", cik="0000000044")
db_session.add_all([
ticker,
SystemSetting(
key="fundamental_data_sec_dolt_cutover_enabled",
value="true",
),
DataImportRun(
source="sec_facts",
status="promoted",
validation_json=json.dumps({
"setup_blocked_ciks": [ticker.cik],
"no_xbrl_ciks": [ticker.cik],
"no_xbrl_filings": [],
}),
started_at=datetime.now(timezone.utc),
),
])
await db_session.flush()
quality = await fundamentals_quality_service.ticker_quality(db_session, "NEWREG")
assert quality.eligible is False
assert quality.code == "no_xbrl_filings"
assert "CIK override" in (quality.message or "")
assert await fundamentals_quality_service.ticker_is_eligible(
db_session, ticker.id
) is False
+78 -4
View File
@@ -6,6 +6,8 @@ from datetime import date, timedelta
import pytest import pytest
from app.models.ohlcv import OHLCVRecord
from app.models.settings import IngestionProgress
from app.models.ticker import Ticker from app.models.ticker import Ticker
from app.providers.protocol import OHLCVData from app.providers.protocol import OHLCVData
from app.services import ingestion_service as svc from app.services import ingestion_service as svc
@@ -18,9 +20,12 @@ async def session():
yield s yield s
async def _add_ticker(session, symbol: str) -> None: async def _add_ticker(session, symbol: str) -> Ticker:
session.add(Ticker(symbol=symbol)) ticker = Ticker(symbol=symbol)
session.add(ticker)
await session.commit() await session.commit()
await session.refresh(ticker)
return ticker
def _bars(symbol: str, n: int) -> list[OHLCVData]: def _bars(symbol: str, n: int) -> list[OHLCVData]:
@@ -51,6 +56,50 @@ async def test_happy_path_ingests_bars(session):
assert result.records_ingested == 3 assert result.records_ingested == 3
async def test_incremental_fetch_overlaps_latest_session_and_updates_partial_bar(session):
"""Once today exists, a live refresh must fetch and overwrite it again."""
ticker = await _add_ticker(session, "LIVE")
today = date.today()
session.add_all([
OHLCVRecord(
ticker_id=ticker.id,
date=today - timedelta(days=i),
open=100.0,
high=101.0,
low=99.0,
close=100.0,
volume=1000,
)
for i in range(200)
])
session.add(IngestionProgress(ticker_id=ticker.id, last_ingested_date=today))
await session.commit()
provider = MockMarketDataProvider(ohlcv_data=[
OHLCVData(
ticker="LIVE",
date=today,
open=100.0,
high=124.0,
low=99.0,
close=123.0,
volume=2000,
)
])
result = await svc.fetch_and_ingest(session, provider, "LIVE")
assert provider.calls == [{
"ticker": "LIVE",
"start_date": today,
"end_date": today,
}]
assert result.status == "complete"
assert result.records_ingested == 1
records = await svc.price_service.query_ohlcv(session, "LIVE", today, today)
assert records[0].close == 123.0
assert records[0].volume == 2000
async def test_empty_fetch_with_existing_history_is_up_to_date(session): async def test_empty_fetch_with_existing_history_is_up_to_date(session):
# Covered ticker, just no new bars in the window → complete, not no_data. # Covered ticker, just no new bars in the window → complete, not no_data.
await _add_ticker(session, "BBB") await _add_ticker(session, "BBB")
@@ -81,9 +130,34 @@ async def test_empty_fetch_with_stale_history_reports_stale(session):
] ]
await svc.fetch_and_ingest(session, MockMarketDataProvider(ohlcv_data=old), "SATS") await svc.fetch_and_ingest(session, MockMarketDataProvider(ohlcv_data=old), "SATS")
result = await svc.fetch_and_ingest(session, MockMarketDataProvider(ohlcv_data=[]), "SATS") # Incremental overlap means Alpaca can keep returning the final historical
# bar. That is still stale: the latest session did not advance.
result = await svc.fetch_and_ingest(
session,
MockMarketDataProvider(ohlcv_data=[old[-1]]),
"SATS",
)
assert result.status == "stale" assert result.status == "stale"
assert result.records_ingested == 0 assert result.records_ingested == 1
assert result.last_date is not None assert result.last_date is not None
assert "renamed" in (result.message or "").lower() or "halted" in (result.message or "").lower() assert "renamed" in (result.message or "").lower() or "halted" in (result.message or "").lower()
async def test_ingest_can_skip_sr_refresh_when_scanner_follows(session, monkeypatch):
await _add_ticker(session, "SCAN")
calls: list[str] = []
async def fake_refresh(db, symbol):
calls.append(symbol)
monkeypatch.setattr(svc, "_refresh_structural_sr", fake_refresh)
result = await svc.fetch_and_ingest(
session,
MockMarketDataProvider(ohlcv_data=_bars("SCAN", 3)),
"SCAN",
refresh_sr=False,
)
assert result.status == "complete"
assert calls == []
+110
View File
@@ -56,6 +56,116 @@ async def test_create_and_list_open(session):
assert row["symbol"] == "AAA" assert row["symbol"] == "AAA"
assert row["status"] == "open" assert row["status"] == "open"
assert row["current_price"] == 110.0 # marked to the latest close assert row["current_price"] == 110.0 # marked to the latest close
assert row["sessions_held"] == 0
assert row["sessions_remaining"] == 30
async def test_list_open_counts_post_entry_sessions_for_max_hold(session):
await svc.set_exit_policy(session, mode="atr_trailing", hold_days=5)
ticker_id = await _seed(session, "COUNT", close=110.0)
trade = await svc.create_trade(
session,
1,
symbol="COUNT",
direction="long",
entry_price=100.0,
shares=10,
stop_loss=95.0,
target=120.0,
)
today = _today()
trade.opened_at = datetime.combine(
today - timedelta(days=5), datetime.min.time(), tzinfo=timezone.utc
)
session.add_all([
OHLCVRecord(
ticker_id=ticker_id,
date=today - timedelta(days=4),
open=101,
high=102,
low=100,
close=101,
volume=1,
),
OHLCVRecord(
ticker_id=ticker_id,
date=today - timedelta(days=2),
open=102,
high=103,
low=101,
close=102,
volume=1,
),
])
await session.commit()
row = (await svc.list_trades(session, 1, status="open"))[0]
# Two added bars plus today's seeded bar; skipped calendar dates do not count.
assert row["sessions_held"] == 3
assert row["sessions_remaining"] == 2
async def test_list_open_exposes_past_max_hold_after_policy_is_shortened(session):
await svc.set_exit_policy(session, mode="time", hold_days=2)
ticker_id = await _seed(session, "OVERDUE", close=110.0)
trade = await svc.create_trade(
session,
1,
symbol="OVERDUE",
direction="long",
entry_price=100.0,
shares=10,
stop_loss=95.0,
target=120.0,
)
today = _today()
trade.opened_at = datetime.combine(
today - timedelta(days=5), datetime.min.time(), tzinfo=timezone.utc
)
session.add_all([
OHLCVRecord(
ticker_id=ticker_id,
date=today - timedelta(days=4),
open=101,
high=102,
low=100,
close=101,
volume=1,
),
OHLCVRecord(
ticker_id=ticker_id,
date=today - timedelta(days=2),
open=102,
high=103,
low=101,
close=102,
volume=1,
),
])
await session.commit()
row = (await svc.list_trades(session, 1, status="open"))[0]
assert row["sessions_held"] == 3
assert row["sessions_remaining"] == -1
async def test_list_open_omits_countdown_without_max_hold_policy(session):
await svc.set_exit_policy(session, mode="trailing")
await _seed(session, "NOHOLD", close=110.0)
await svc.create_trade(
session,
1,
symbol="NOHOLD",
direction="long",
entry_price=100.0,
shares=10,
stop_loss=95.0,
target=120.0,
)
row = (await svc.list_trades(session, 1, status="open"))[0]
assert row["sessions_held"] is None
assert row["sessions_remaining"] is None
async def test_create_trade_enforces_post_stop_gate_reset_at_service_boundary(session): async def test_create_trade_enforces_post_stop_gate_reset_at_service_boundary(session):
@@ -1,285 +0,0 @@
"""Regression: scanner must not headline the most distant (max raw R:R) level.
Historical bug: provisional candidate pick used max R:R / quality only. Production
headline is probability-based primary after enhance_trade_setup — near levels
with real reach-probability beat far lotteries.
**Validates: Requirements 1.1, 1.3, 1.4, 2.1, 2.3, 2.4**
"""
from __future__ import annotations
from datetime import date, timedelta
import pytest
from hypothesis import given, settings, HealthCheck, strategies as st
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.ohlcv import OHLCVRecord
from app.models.sr_level import SRLevel
from app.models.ticker import Ticker
from app.services.rr_scanner_service import scan_ticker
# ---------------------------------------------------------------------------
# Session fixture that allows scan_ticker to commit
# ---------------------------------------------------------------------------
# The default db_session fixture wraps in session.begin() which conflicts
# with scan_ticker's internal commit(). We use a plain session instead.
@pytest.fixture
async def scan_session() -> AsyncSession:
"""Provide a DB session compatible with scan_ticker (which commits)."""
from tests.conftest import _test_session_factory
async with _test_session_factory() as session:
yield session
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_ohlcv_bars(
ticker_id: int,
num_bars: int = 20,
base_close: float = 100.0,
) -> list[OHLCVRecord]:
"""Generate realistic OHLCV bars with small daily variation.
Produces bars where close ≈ base_close, with enough range for ATR
computation (needs >= 15 bars). The ATR will be roughly 2.0.
"""
bars: list[OHLCVRecord] = []
start = date(2024, 1, 1)
for i in range(num_bars):
close = base_close + (i % 3 - 1) * 0.5 # oscillate ±0.5
bars.append(OHLCVRecord(
ticker_id=ticker_id,
date=start + timedelta(days=i),
open=close - 0.3,
high=close + 1.0,
low=close - 1.0,
close=close,
volume=100_000,
))
return bars
# ---------------------------------------------------------------------------
# Deterministic test: strong-near vs weak-far (long setup)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_long_prefers_strong_near_over_weak_far(scan_session: AsyncSession):
"""With a strong nearby resistance and a weak distant resistance, the
probability primary should be the nearby level — NOT the far lottery.
"""
ticker = Ticker(symbol="EXPLR")
scan_session.add(ticker)
await scan_session.flush()
# 20 bars closing around 100
bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0)
scan_session.add_all(bars)
# With ATR=2.0 and multiplier=1.5, risk=3.0.
# R:R threshold=1.5 → min reward=4.5 → min target=104.5
# Strong nearby resistance: price=105, strength=90 (R:R≈1.67, quality≈0.66)
near_level = SRLevel(
ticker_id=ticker.id,
price_level=105.0,
type="resistance",
strength=90,
detection_method="volume_profile",
)
# Weak distant resistance: price=130, strength=5 (R:R=10, quality≈0.58)
far_level = SRLevel(
ticker_id=ticker.id,
price_level=130.0,
type="resistance",
strength=5,
detection_method="volume_profile",
)
scan_session.add_all([near_level, far_level])
await scan_session.flush()
setups = await scan_ticker(
scan_session,
"EXPLR",
rr_threshold=1.5,
gate_levels_override=[near_level, far_level],
)
long_setups = [s for s in setups if s.direction == "long"]
assert len(long_setups) == 1, "Expected exactly one long setup"
selected_target = long_setups[0].target
# The scanner must NOT pick the most distant level (130)
assert selected_target != pytest.approx(130.0, abs=0.01), (
"Bug: scanner picked the weak distant level (130) instead of the "
"strong nearby level (105)"
)
# Probability primary should pick the strong nearby level
assert selected_target == pytest.approx(105.0, abs=0.01)
primaries = [t for t in long_setups[0].targets if t.get("is_primary")]
assert len(primaries) == 1
assert primaries[0]["price"] == pytest.approx(105.0, abs=0.01)
# ---------------------------------------------------------------------------
# Deterministic test: strong-near vs weak-far (short setup)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_short_prefers_strong_near_over_weak_far(scan_session: AsyncSession):
"""Short-side mirror: strong nearby support should be preferred over
weak distant support.
"""
ticker = Ticker(symbol="EXPLS")
scan_session.add(ticker)
await scan_session.flush()
bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0)
scan_session.add_all(bars)
# With ATR=2.0 and multiplier=1.5, risk=3.0.
# R:R threshold=1.5 → min reward=4.5 → min target below 95.5
# Strong nearby support: price=95, strength=85 (R:R≈1.67, quality≈0.64)
near_level = SRLevel(
ticker_id=ticker.id,
price_level=95.0,
type="support",
strength=85,
detection_method="pivot_point",
)
# Weak distant support: price=70, strength=5 (R:R=10, quality≈0.58)
far_level = SRLevel(
ticker_id=ticker.id,
price_level=70.0,
type="support",
strength=5,
detection_method="pivot_point",
)
scan_session.add_all([near_level, far_level])
await scan_session.flush()
setups = await scan_ticker(
scan_session,
"EXPLS",
rr_threshold=1.5,
gate_levels_override=[near_level, far_level],
)
short_setups = [s for s in setups if s.direction == "short"]
assert len(short_setups) == 1, "Expected exactly one short setup"
selected_target = short_setups[0].target
assert selected_target != pytest.approx(70.0, abs=0.01), (
"Bug: scanner picked the weak distant level (70) instead of the "
"strong nearby level (95)"
)
assert selected_target == pytest.approx(95.0, abs=0.01)
# ---------------------------------------------------------------------------
# Hypothesis property test: selection is NOT always the most distant level
# ---------------------------------------------------------------------------
@st.composite
def strong_near_weak_far_pair(draw: st.DrawFn) -> dict:
"""Generate a (strong-near, weak-far) resistance pair above entry=100.
Guarantees:
- near_price < far_price (both above entry)
- near_strength >> far_strength
- Both meet the R:R threshold of 1.5 given typical ATR ≈ 2 → risk ≈ 3
"""
# Near level: 515 above entry (R:R ≈ 1.75.0 with risk≈3)
near_dist = draw(st.floats(min_value=5.0, max_value=15.0))
near_strength = draw(st.integers(min_value=70, max_value=100))
# Far level: 2560 above entry (R:R ≈ 8.320 with risk≈3)
far_dist = draw(st.floats(min_value=25.0, max_value=60.0))
far_strength = draw(st.integers(min_value=1, max_value=15))
return {
"near_price": 100.0 + near_dist,
"near_strength": near_strength,
"far_price": 100.0 + far_dist,
"far_strength": far_strength,
}
@pytest.mark.asyncio
@given(pair=strong_near_weak_far_pair())
@settings(
max_examples=15,
deadline=None,
suppress_health_check=[HealthCheck.function_scoped_fixture],
)
async def test_property_scanner_does_not_always_pick_most_distant(
pair: dict,
scan_session: AsyncSession,
):
"""**Validates: Requirements 1.1, 1.3, 1.4, 2.1, 2.3, 2.4**
Property: when a strong nearby resistance exists alongside a weak distant
resistance, the scanner does NOT always select the most distant level.
On unfixed code this would fail for every example because max-R:R always
picks the farthest level.
"""
from tests.conftest import _test_engine, _test_session_factory
# Each hypothesis example needs a fresh DB state
async with _test_engine.begin() as conn:
from app.database import Base
await conn.run_sync(Base.metadata.drop_all)
await conn.run_sync(Base.metadata.create_all)
async with _test_session_factory() as session:
ticker = Ticker(symbol="PROP")
session.add(ticker)
await session.flush()
bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0)
session.add_all(bars)
near_level = SRLevel(
ticker_id=ticker.id,
price_level=pair["near_price"],
type="resistance",
strength=pair["near_strength"],
detection_method="volume_profile",
)
far_level = SRLevel(
ticker_id=ticker.id,
price_level=pair["far_price"],
type="resistance",
strength=pair["far_strength"],
detection_method="volume_profile",
)
session.add_all([near_level, far_level])
await session.commit()
setups = await scan_ticker(
session,
"PROP",
rr_threshold=1.5,
gate_levels_override=[near_level, far_level],
)
long_setups = [s for s in setups if s.direction == "long"]
assert len(long_setups) == 1, "Expected exactly one long setup"
selected_target = long_setups[0].target
most_distant = round(pair["far_price"], 4)
# The fixed scanner should prefer the strong nearby level, not the
# most distant weak one.
assert selected_target != pytest.approx(most_distant, abs=0.01), (
f"Bug: scanner picked the most distant level ({most_distant}) "
f"with strength={pair['far_strength']} over the nearby level "
f"({round(pair['near_price'], 4)}) with strength={pair['near_strength']}"
)
-375
View File
@@ -1,375 +0,0 @@
"""Fix-checking tests for R:R scanner probability-based primary selection.
Verify that after enhance_trade_setup the headline target is the most likely
worthwhile primary (R:R + probability floors), for both long and short setups.
The pre-enhance quality loop only seeds a provisional target.
**Validates: Requirements 2.1, 2.2, 2.3, 2.4**
"""
from __future__ import annotations
from datetime import date, timedelta
import pytest
from hypothesis import given, settings, HealthCheck, strategies as st
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.ohlcv import OHLCVRecord
from app.models.sr_level import SRLevel
from app.models.ticker import Ticker
from app.services.rr_scanner_service import scan_ticker
def _assert_primary_is_most_likely_worthwhile(setup) -> None:
"""Headline = starred primary = max(probability, rr) among floor-clearing targets."""
targets = setup.targets
assert targets, "expected generated targets"
primaries = [t for t in targets if t.get("is_primary")]
assert len(primaries) == 1, "exactly one primary target expected"
primary = primaries[0]
assert setup.target == pytest.approx(primary["price"], abs=0.01)
# Mirrors recommendation_service._select_primary_target floors.
worthwhile = [
t for t in targets
if float(t["rr_ratio"]) >= 1.5 and float(t["probability"]) >= 20.0
]
pool = worthwhile or targets
best = max(pool, key=lambda t: (t["probability"], t["rr_ratio"]))
assert primary["price"] == pytest.approx(best["price"], abs=0.01)
# ---------------------------------------------------------------------------
# Session fixture (plain session, not wrapped in begin())
# ---------------------------------------------------------------------------
@pytest.fixture
async def scan_session() -> AsyncSession:
"""Provide a DB session compatible with scan_ticker (which commits)."""
from tests.conftest import _test_session_factory
async with _test_session_factory() as session:
yield session
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_ohlcv_bars(
ticker_id: int,
num_bars: int = 20,
base_close: float = 100.0,
) -> list[OHLCVRecord]:
"""Generate OHLCV bars closing around base_close with ATR ≈ 2.0."""
bars: list[OHLCVRecord] = []
start = date(2024, 1, 1)
for i in range(num_bars):
close = base_close + (i % 3 - 1) * 0.5 # oscillate ±0.5
bars.append(OHLCVRecord(
ticker_id=ticker_id,
date=start + timedelta(days=i),
open=close - 0.3,
high=close + 1.0,
low=close - 1.0,
close=close,
volume=100_000,
))
return bars
# ---------------------------------------------------------------------------
# Hypothesis strategy: multiple resistance levels above entry for longs
# ---------------------------------------------------------------------------
@st.composite
def long_candidate_levels(draw: st.DrawFn) -> list[dict]:
"""Generate 2-5 resistance levels above entry_price=100.
All levels meet the R:R threshold of 1.5 given ATR≈2, risk≈3,
so min reward=4.5, min target=104.5.
"""
num_levels = draw(st.integers(min_value=2, max_value=5))
levels = []
for _ in range(num_levels):
# Distance from entry: 5 to 50 (all above 4.5 threshold)
distance = draw(st.floats(min_value=5.0, max_value=50.0))
strength = draw(st.integers(min_value=0, max_value=100))
levels.append({
"price": 100.0 + distance,
"strength": strength,
})
return levels
@st.composite
def short_candidate_levels(draw: st.DrawFn) -> list[dict]:
"""Generate 2-5 support levels below entry_price=100.
All levels meet the R:R threshold of 1.5 given ATR≈2, risk≈3,
so min reward=4.5, max target=95.5.
"""
num_levels = draw(st.integers(min_value=2, max_value=5))
levels = []
for _ in range(num_levels):
# Distance below entry: 5 to 50 (all above 4.5 threshold)
distance = draw(st.floats(min_value=5.0, max_value=50.0))
strength = draw(st.integers(min_value=0, max_value=100))
levels.append({
"price": 100.0 - distance,
"strength": strength,
})
return levels
# ---------------------------------------------------------------------------
# Property test: long setup selects probability-based primary
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
@given(levels=long_candidate_levels())
@settings(
max_examples=20,
deadline=None,
suppress_health_check=[HealthCheck.function_scoped_fixture],
)
async def test_property_long_selects_probability_primary(
levels: list[dict],
scan_session: AsyncSession,
):
"""**Validates: Requirements 2.1, 2.3, 2.4**
Property: when multiple resistance levels meet the R:R threshold,
the headline after enhance is the probability-based primary.
"""
from tests.conftest import _test_engine, _test_session_factory
from app.database import Base
# Fresh DB state per hypothesis example
async with _test_engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await conn.run_sync(Base.metadata.create_all)
async with _test_session_factory() as session:
ticker = Ticker(symbol="FIXL")
session.add(ticker)
await session.flush()
bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0)
session.add_all(bars)
sr_levels = []
for lv in levels:
sr_levels.append(SRLevel(
ticker_id=ticker.id,
price_level=lv["price"],
type="resistance",
strength=lv["strength"],
detection_method="volume_profile",
))
session.add_all(sr_levels)
await session.commit()
setups = await scan_ticker(
session,
"FIXL",
rr_threshold=1.5,
gate_levels_override=sr_levels,
)
long_setups = [s for s in setups if s.direction == "long"]
assert len(long_setups) == 1, "Expected exactly one long setup"
_assert_primary_is_most_likely_worthwhile(long_setups[0])
# ---------------------------------------------------------------------------
# Property test: short setup selects probability-based primary
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
@given(levels=short_candidate_levels())
@settings(
max_examples=20,
deadline=None,
suppress_health_check=[HealthCheck.function_scoped_fixture],
)
async def test_property_short_selects_probability_primary(
levels: list[dict],
scan_session: AsyncSession,
):
"""**Validates: Requirements 2.2, 2.3, 2.4**
Property: when multiple support levels meet the R:R threshold,
the headline after enhance is the probability-based primary.
"""
from tests.conftest import _test_engine, _test_session_factory
from app.database import Base
# Fresh DB state per hypothesis example
async with _test_engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await conn.run_sync(Base.metadata.create_all)
async with _test_session_factory() as session:
ticker = Ticker(symbol="FIXS")
session.add(ticker)
await session.flush()
bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0)
session.add_all(bars)
sr_levels = []
for lv in levels:
sr_levels.append(SRLevel(
ticker_id=ticker.id,
price_level=lv["price"],
type="support",
strength=lv["strength"],
detection_method="pivot_point",
))
session.add_all(sr_levels)
await session.commit()
setups = await scan_ticker(
session,
"FIXS",
rr_threshold=1.5,
gate_levels_override=sr_levels,
)
short_setups = [s for s in setups if s.direction == "short"]
assert len(short_setups) == 1, "Expected exactly one short setup"
_assert_primary_is_most_likely_worthwhile(short_setups[0])
# ---------------------------------------------------------------------------
# Deterministic test: 3 levels with known quality scores (long)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_deterministic_long_three_levels(scan_session: AsyncSession):
"""**Validates: Requirements 2.1, 2.3, 2.4**
Concrete example with 3 resistance levels of known quality scores.
Entry=100, ATR≈2, risk≈3.
Level A: price=105, strength=90 → rr=5/3≈1.67, dist=5
quality = 0.35*(1.67/10) + 0.35*(90/100) + 0.30*(1-5/100)
= 0.35*0.167 + 0.35*0.9 + 0.30*0.95
= 0.0585 + 0.315 + 0.285 = 0.6585
Level B: price=112, strength=50 → rr=12/3=4.0, dist=12
quality = 0.35*(4/10) + 0.35*(50/100) + 0.30*(1-12/100)
= 0.35*0.4 + 0.35*0.5 + 0.30*0.88
= 0.14 + 0.175 + 0.264 = 0.579
Level C: price=130, strength=10 → rr=30/3=10.0, dist=30
quality = 0.35*(10/10) + 0.35*(10/100) + 0.30*(1-30/100)
= 0.35*1.0 + 0.35*0.1 + 0.30*0.7
= 0.35 + 0.035 + 0.21 = 0.595
Expected winner: Level A (quality=0.6585)
"""
ticker = Ticker(symbol="DET3L")
scan_session.add(ticker)
await scan_session.flush()
bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0)
scan_session.add_all(bars)
level_a = SRLevel(
ticker_id=ticker.id, price_level=105.0, type="resistance",
strength=90, detection_method="volume_profile",
)
level_b = SRLevel(
ticker_id=ticker.id, price_level=112.0, type="resistance",
strength=50, detection_method="volume_profile",
)
level_c = SRLevel(
ticker_id=ticker.id, price_level=130.0, type="resistance",
strength=10, detection_method="volume_profile",
)
scan_session.add_all([level_a, level_b, level_c])
await scan_session.flush()
setups = await scan_ticker(
scan_session,
"DET3L",
rr_threshold=1.5,
gate_levels_override=[level_a, level_b, level_c],
)
long_setups = [s for s in setups if s.direction == "long"]
assert len(long_setups) == 1, "Expected exactly one long setup"
_assert_primary_is_most_likely_worthwhile(long_setups[0])
# Near/strong level A wins on reach-probability over far lottery C.
assert long_setups[0].target == pytest.approx(105.0, abs=0.01), (
f"Expected primary=105.0 (near, high reach-prob), got {long_setups[0].target}"
)
# ---------------------------------------------------------------------------
# Deterministic test: 3 levels with known quality scores (short)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_deterministic_short_three_levels(scan_session: AsyncSession):
"""**Validates: Requirements 2.2, 2.3, 2.4**
Concrete example with 3 support levels of known quality scores.
Entry=100, ATR≈2, risk≈3.
Level A: price=95, strength=85 → rr=5/3≈1.67, dist=5
quality = 0.35*(1.67/10) + 0.35*(85/100) + 0.30*(1-5/100)
= 0.0585 + 0.2975 + 0.285 = 0.641
Level B: price=88, strength=45 → rr=12/3=4.0, dist=12
quality = 0.35*(4/10) + 0.35*(45/100) + 0.30*(1-12/100)
= 0.14 + 0.1575 + 0.264 = 0.5615
Level C: price=70, strength=8 → rr=30/3=10.0, dist=30
quality = 0.35*(10/10) + 0.35*(8/100) + 0.30*(1-30/100)
= 0.35 + 0.028 + 0.21 = 0.588
Expected winner: Level A (quality=0.641)
"""
ticker = Ticker(symbol="DET3S")
scan_session.add(ticker)
await scan_session.flush()
bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0)
scan_session.add_all(bars)
level_a = SRLevel(
ticker_id=ticker.id, price_level=95.0, type="support",
strength=85, detection_method="pivot_point",
)
level_b = SRLevel(
ticker_id=ticker.id, price_level=88.0, type="support",
strength=45, detection_method="pivot_point",
)
level_c = SRLevel(
ticker_id=ticker.id, price_level=70.0, type="support",
strength=8, detection_method="pivot_point",
)
scan_session.add_all([level_a, level_b, level_c])
await scan_session.flush()
setups = await scan_ticker(
scan_session,
"DET3S",
rr_threshold=1.5,
gate_levels_override=[level_a, level_b, level_c],
)
short_setups = [s for s in setups if s.direction == "short"]
assert len(short_setups) == 1, "Expected exactly one short setup"
_assert_primary_is_most_likely_worthwhile(short_setups[0])
assert short_setups[0].target == pytest.approx(95.0, abs=0.01), (
f"Expected primary=95.0 (near, high reach-prob), got {short_setups[0].target}"
)
@@ -23,6 +23,8 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.models.ohlcv import OHLCVRecord from app.models.ohlcv import OHLCVRecord
from app.models.paper_trade import PaperTrade from app.models.paper_trade import PaperTrade
from app.models.signal_context_snapshot import SignalContextSnapshot from app.models.signal_context_snapshot import SignalContextSnapshot
from app.models.sec_filing_gap import SecFilingGap
from app.models.settings import SystemSetting
from app.models.sr_level import SRLevel from app.models.sr_level import SRLevel
from app.models.ticker import Ticker from app.models.ticker import Ticker
from app.models.trade_setup import TradeSetup from app.models.trade_setup import TradeSetup
@@ -513,6 +515,45 @@ async def test_get_trade_setups_excludes_stale_rows(db_session: AsyncSession):
assert stale_rows == [] assert stale_rows == []
@pytest.mark.asyncio
async def test_get_trade_setups_hides_active_sec_filing_gap(
db_session: AsyncSession,
):
now = datetime.now(timezone.utc)
ticker = Ticker(symbol="SECWAIT", cik="0000000042")
db_session.add(ticker)
await db_session.flush()
db_session.add_all([
SystemSetting(
key="fundamental_data_sec_dolt_cutover_enabled",
value="true",
),
SecFilingGap(
cik=ticker.cik,
accession="0000000042-26-000001",
form="10-Q",
index_date=date.today(),
reason="not_in_companyfacts",
first_seen_at=now,
last_attempted_at=now,
),
TradeSetup(
ticker_id=ticker.id,
direction="long",
entry_price=100.0,
stop_loss=97.0,
target=109.0,
rr_ratio=3.0,
composite_score=70.0,
confidence_score=80.0,
detected_at=now,
),
])
await db_session.flush()
assert await get_trade_setups(db_session, symbol="SECWAIT") == []
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_get_trade_setups_can_exclude_tickers_with_open_paper_trades( async def test_get_trade_setups_can_exclude_tickers_with_open_paper_trades(
db_session: AsyncSession, db_session: AsyncSession,
+58
View File
@@ -108,3 +108,61 @@ async def test_scan_error_does_not_stop_later_tickers(session, monkeypatch):
await rr_scanner_service.scan_all_tickers(session) await rr_scanner_service.scan_all_tickers(session)
assert scanned == ["AAA", "BBB"] assert scanned == ["AAA", "BBB"]
async def test_scan_skips_ticker_with_incomplete_sec_fundamentals(
session, monkeypatch
):
ticker = Ticker(symbol="BLOCKED", cik="0000000001")
session.add(ticker)
await session.commit()
async def _blocked(db):
return {ticker.id}
async def _unexpected_scan(*args, **kwargs):
raise AssertionError("fundamentals-incomplete ticker was scanned")
monkeypatch.setattr(
rr_scanner_service.fundamentals_quality_service,
"blocked_ticker_ids",
_blocked,
)
monkeypatch.setattr(rr_scanner_service, "scan_ticker", _unexpected_scan)
assert await rr_scanner_service.scan_all_tickers(session) == []
async def test_scan_quality_failure_blocks_closed_and_emits_event(
session, monkeypatch
):
session.add(Ticker(symbol="BLOCKED"))
await session.commit()
async def _boom(db):
raise ValueError("bad quality metadata")
async def _unexpected_scan(*args, **kwargs):
raise AssertionError("ticker was scanned without a quality decision")
events: list[dict] = []
async def _capture_event(**kwargs):
events.append(kwargs)
monkeypatch.setattr(
rr_scanner_service.fundamentals_quality_service,
"blocked_ticker_ids",
_boom,
)
monkeypatch.setattr(rr_scanner_service, "scan_ticker", _unexpected_scan)
monkeypatch.setattr(
rr_scanner_service.system_event_service,
"log_event_standalone",
_capture_event,
)
assert await rr_scanner_service.scan_all_tickers(session) == []
assert [event["code"] for event in events] == [
"fundamentals_quality_unavailable"
]
+10
View File
@@ -5,6 +5,8 @@ from types import SimpleNamespace
import pytest import pytest
from app.scheduler import ( from app.scheduler import (
_DAILY_PIPELINE_STEPS,
_NEAR_CLOSE_PIPELINE_STEPS,
_consume_backtest_options, _consume_backtest_options,
_consume_backtest_target_model, _consume_backtest_target_model,
_parse_frequency, _parse_frequency,
@@ -43,6 +45,14 @@ def test_manual_backtest_options_are_one_shot_and_default_back_to_weekly():
assert _consume_backtest_options() == ("production_gtl", "weekly") assert _consume_backtest_options() == ("production_gtl", "weekly")
def test_only_near_close_fetch_skips_redundant_sr_refresh():
assert dict(_DAILY_PIPELINE_STEPS)["data_collector"] == "collect_ohlcv"
assert (
dict(_NEAR_CLOSE_PIPELINE_STEPS)["data_collector"]
== "collect_ohlcv_for_scan"
)
class TestParseFrequency: class TestParseFrequency:
def test_hourly(self): def test_hourly(self):
assert _parse_frequency("hourly") == {"hours": 1} assert _parse_frequency("hourly") == {"hours": 1}
+324 -1
View File
@@ -15,6 +15,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn
from app.database import Base from app.database import Base
import app.models # noqa: F401 import app.models # noqa: F401
from app.models.fundamental_snapshot import FundamentalSnapshot from app.models.fundamental_snapshot import FundamentalSnapshot
from app.models.sec_filing_gap import SecFilingGap
from app.models.system_event import SystemEvent from app.models.system_event import SystemEvent
from app.models.ticker import Ticker from app.models.ticker import Ticker
from app.services.data_import import ( from app.services.data_import import (
@@ -374,7 +375,7 @@ async def test_recovers_facts_misfiled_under_coregistrant(engine):
# Stamped to the issuer that filed, NOT the co-registrant whose file it came from. # Stamped to the issuer that filed, NOT the co-registrant whose file it came from.
assert q2.cik == "0000320193" assert q2.cik == "0000320193"
assert q2.revenue == 254940 and q2.shares_outstanding == 14687 assert q2.revenue == 254940 and q2.shares_outstanding == 14687
assert "coregistrant_recovery" in codes assert "coregistrant_recovery" not in codes
async def test_coregistrant_recovery_rejects_discontinuous_share_count(engine): async def test_coregistrant_recovery_rejects_discontinuous_share_count(engine):
@@ -425,11 +426,333 @@ async def test_unresolved_filing_stops_blocking_after_retry_window(engine):
assert run.source_max_date == date(2026, 5, 2) # and the index advances assert run.source_max_date == date(2026, 5, 2) # and the index advances
summary = json.loads(run.validation_json or "{}") summary = json.loads(run.validation_json or "{}")
assert summary["missing_xbrl_count"] == 1 and summary["missing_xbrl_blocking"] == 0 assert summary["missing_xbrl_count"] == 1 and summary["missing_xbrl_blocking"] == 0
assert await _count(factory, SecFilingGap) == 1
async with factory() as s: async with factory() as s:
events = (await s.execute(select(SystemEvent))).scalars().all() events = (await s.execute(select(SystemEvent))).scalars().all()
unresolved = [e for e in events if e.code == "unresolved_filing"] unresolved = [e for e in events if e.code == "unresolved_filing"]
assert len(unresolved) == 1 and "GHOST" in unresolved[0].message assert len(unresolved) == 1 and "GHOST" in unresolved[0].message
assert "automatic SEC retry" in unresolved[0].message
# The next scheduled run retries even though the SEC daily-index revision
# has not changed. Company Facts is a separate SEC product and may catch up
# independently, so the generic revision no-op must not suppress this work.
still_missing_run = await run_import(
_importer(incr, today=date(2026, 5, 11)),
engine=engine,
)
assert still_missing_run.status == STATUS_PROMOTED
assert still_missing_run.revision is None
assert await _count(factory, SecFilingGap) == 1
# A later normal scheduled import retries only the queued issuer. Once SEC
# publishes the accession in Company Facts, it is inserted and unblocked
# without a full-universe reparse or operator action.
cf_ghost = _rev("2025-09-28", "2026-03-28", 254940, 2026, "Q2", "GHOST")
sh_ghost = _shares("2026-04-17", 14687, "GHOST", 2026, "Q2")
healed = FakeSecClient(
tickers={"AAPL": 320193},
companyfacts={
320193: _companyfacts(
[CF_K, CF_Q1, cf_ghost],
[SH_K, SH_Q1, sh_ghost],
)
},
submissions={
320193: _submissions(SUB_FILINGS + [
_filing(
"GHOST",
"10-Q",
"2026-03-28",
"2026-05-01",
"2026-05-01T10:01:00.000Z",
)
])
},
latest_index=date(2026, 5, 2),
)
healed_run = await run_import(
_importer(healed, today=date(2026, 5, 12)),
engine=engine,
)
assert healed_run.status == STATUS_PROMOTED
assert await _count(factory, FundamentalSnapshot) == 3
assert await _count(factory, SecFilingGap) == 0
async def test_queued_gap_without_index_date_retries_without_wedging_and_escalates_once(
engine,
):
factory = _factory(engine)
await _seed(factory, ["AAPL"])
backfill = FakeSecClient(
tickers={"AAPL": 320193},
companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])},
submissions={320193: _submissions(SUB_FILINGS)},
latest_index=date(2026, 1, 31),
)
await run_import(_importer(backfill), engine=engine)
old = datetime(2026, 4, 1, tzinfo=timezone.utc)
async with factory() as db:
db.add(SecFilingGap(
cik="0000320193",
accession="DATELESS",
form="10-Q",
index_date=None,
reason="not_in_companyfacts",
first_seen_at=old,
last_attempted_at=old,
))
await db.commit()
missing = FakeSecClient(
tickers={"AAPL": 320193},
companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])},
submissions={320193: _submissions(SUB_FILINGS + [
_filing(
"DATELESS",
"10-Q",
"2026-03-28",
"2026-05-01",
"2026-05-01T10:01:00.000Z",
)
])},
latest_index=date(2026, 1, 31),
)
first = await run_import(
_importer(missing, today=date(2026, 5, 20)), engine=engine
)
second = await run_import(
_importer(missing, today=date(2026, 5, 21)), engine=engine
)
assert first.status == STATUS_PROMOTED
assert second.status == STATUS_PROMOTED
async with factory() as db:
gap = (await db.execute(select(SecFilingGap))).scalar_one()
events = (
await db.execute(
select(SystemEvent).where(SystemEvent.code == "filing_gap_aged")
)
).scalars().all()
assert gap.escalated_at is not None
assert len(events) == 1
async def test_queued_filing_reclassified_non_xbrl_is_removed(engine):
factory = _factory(engine)
await _seed(factory, ["AAPL"])
backfill = FakeSecClient(
tickers={"AAPL": 320193},
companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])},
submissions={320193: _submissions(SUB_FILINGS)},
latest_index=date(2026, 1, 31),
)
await run_import(_importer(backfill), engine=engine)
now = datetime.now(timezone.utc)
async with factory() as db:
db.add(SecFilingGap(
cik="0000320193",
accession="NONX",
form="10-Q/A",
index_date=date(2026, 5, 1),
reason="not_in_companyfacts",
first_seen_at=now,
last_attempted_at=now,
))
await db.commit()
client = FakeSecClient(
tickers={"AAPL": 320193},
companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])},
submissions={320193: _submissions(SUB_FILINGS + [
_filing(
"NONX",
"10-Q/A",
"2026-03-28",
"2026-05-01",
"2026-05-01T10:01:00.000Z",
is_xbrl=False,
)
])},
latest_index=date(2026, 1, 31),
)
run = await run_import(_importer(client, today=date(2026, 5, 20)), engine=engine)
assert run.status == STATUS_PROMOTED
assert await _count(factory, SecFilingGap) == 0
async def test_queued_parser_skip_stays_blocked_with_actionable_reason(
engine, monkeypatch
):
from app.services.sec_facts_parser import ParseResult
factory = _factory(engine)
await _seed(factory, ["AAPL"])
backfill = FakeSecClient(
tickers={"AAPL": 320193},
companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])},
submissions={320193: _submissions(SUB_FILINGS)},
latest_index=date(2026, 1, 31),
)
await run_import(_importer(backfill), engine=engine)
now = datetime.now(timezone.utc)
async with factory() as db:
db.add(SecFilingGap(
cik="0000320193",
accession="BADPARSE",
form="10-Q",
index_date=date(2026, 5, 1),
reason="not_in_companyfacts",
first_seen_at=now,
last_attempted_at=now,
))
await db.commit()
bad_fact = _rev(
"2025-09-28", "2026-03-28", 254940, 2026, "Q2", "BADPARSE"
)
bad_share = _shares("2026-04-17", 14687, "BADPARSE", 2026, "Q2")
client = FakeSecClient(
tickers={"AAPL": 320193},
companyfacts={
320193: _companyfacts(
[CF_K, CF_Q1, bad_fact], [SH_K, SH_Q1, bad_share]
)
},
submissions={320193: _submissions(SUB_FILINGS + [
_filing(
"BADPARSE",
"10-Q",
"2026-03-28",
"2026-05-01",
"2026-05-01T10:01:00.000Z",
)
])},
latest_index=date(2026, 1, 31),
)
def skip_parse(*args, **kwargs):
return ParseResult(skipped_filings=[{
"accession": "BADPARSE",
"reason": "unparseable",
}])
monkeypatch.setattr("app.services.sec_facts_parser.parse_snapshots", skip_parse)
run = await run_import(_importer(client, today=date(2026, 5, 20)), engine=engine)
assert run.status == STATUS_PROMOTED
async with factory() as db:
gap = (await db.execute(select(SecFilingGap))).scalar_one()
assert gap.reason == "parser_unusable"
async def test_new_parser_skip_gets_grace_then_enters_retry_queue(
engine, monkeypatch
):
from app.services.sec_facts_parser import ParseResult
factory = _factory(engine)
await _seed(factory, ["AAPL"])
backfill = FakeSecClient(
tickers={"AAPL": 320193},
companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])},
submissions={320193: _submissions(SUB_FILINGS)},
latest_index=date(2026, 1, 31),
)
await run_import(_importer(backfill), engine=engine)
bad_fact = _rev(
"2025-09-28", "2026-03-28", 254940, 2026, "Q2", "NEWBAD"
)
bad_share = _shares("2026-04-17", 14687, "NEWBAD", 2026, "Q2")
client = FakeSecClient(
tickers={"AAPL": 320193},
companyfacts={
320193: _companyfacts(
[CF_K, CF_Q1, bad_fact], [SH_K, SH_Q1, bad_share]
)
},
submissions={320193: _submissions(SUB_FILINGS + [
_filing(
"NEWBAD",
"10-Q",
"2026-03-28",
"2026-05-01",
"2026-05-01T10:01:00.000Z",
)
])},
latest_index=date(2026, 5, 2),
daily={
date(2026, 5, 1): [{
"form": "10-Q",
"cik": 320193,
"accession": "NEWBAD",
}]
},
)
def skip_parse(*args, **kwargs):
return ParseResult(skipped_filings=[{
"accession": "NEWBAD",
"reason": "unparseable",
}])
monkeypatch.setattr("app.services.sec_facts_parser.parse_snapshots", skip_parse)
young = await run_import(
_importer(client, today=date(2026, 5, 3)), engine=engine
)
assert young.status == STATUS_DEFERRED
assert "parser_unusable" in (young.error_details or "")
assert await _count(factory, SecFilingGap) == 0
aged = await run_import(
_importer(client, today=date(2026, 5, 5)), engine=engine
)
assert aged.status == STATUS_PROMOTED
async with factory() as db:
gap = (await db.execute(select(SecFilingGap))).scalar_one()
assert gap.accession == "NEWBAD"
assert gap.reason == "parser_unusable"
async def test_validation_caps_details_but_keeps_complete_blocked_cik_set():
importer = SecFundamentalsImporter(today=date(2026, 5, 20))
importer._latest_index_date = date(2026, 5, 19)
staged = StagedFundamentals(
resolved=ResolvedUniverse(),
missing_xbrl=[
{
"cik": f"{i:010d}",
"accession": f"MISS-{i}",
"form": "10-Q",
"index_date": date(2026, 5, 1),
"age_days": 19,
"reason": "not_in_companyfacts",
}
for i in range(60)
],
no_xbrl_filings=[
{"cik": f"{i + 100:010d}", "name": f"New {i}"}
for i in range(60)
],
recovered=[
{"cik": f"{i:010d}", "accession": f"REC-{i}", "source_cik": "1"}
for i in range(60)
],
)
result = await importer.validate(None, staged)
assert len(result.summary["missing_xbrl"]) == 50
assert len(result.summary["no_xbrl_filings"]) == 50
assert len(result.summary["no_xbrl_ciks"]) == 60
assert len(result.summary["recovered_from_coregistrant"]) == 50
assert len(result.summary["setup_blocked_ciks"]) == 120
async def test_non_xbrl_amendment_skipped_not_failed(engine): async def test_non_xbrl_amendment_skipped_not_failed(engine):