499 lines
17 KiB
Python
499 lines
17 KiB
Python
"""Read-only A5 comparison of legacy and SEC/Dolt fundamental inputs.
|
|
|
|
The report deliberately does not write ``fundamental_data`` or score tables.
|
|
It reconstructs the current legacy and candidate fundamental scores, projects
|
|
their composite-score/rank effect with the active weights, and archives a
|
|
timestamped JSON + CSV bundle for explicit human approval.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import io
|
|
import json
|
|
import math
|
|
import os
|
|
import statistics
|
|
from datetime import date, datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from sqlalchemy import select, text
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.data_import_run import DataImportRun
|
|
from app.models.fundamental import FundamentalData
|
|
from app.services import fundamentals_candidate_service as candidate_service
|
|
|
|
REPORT_VERSION = 1
|
|
APPROVAL_STATUS = "pending_explicit_approval"
|
|
FIELD_KEYS = ("pe_ratio", "revenue_growth", "earnings_surprise")
|
|
MIN_SCORE_METRICS = 2
|
|
|
|
# Materiality is a review aid, never an automatic cutover verdict. Definition
|
|
# changes remain visible even when a delta falls inside these bands.
|
|
FIELD_TOLERANCES = {
|
|
"pe_ratio": {"absolute": 1.0, "relative_pct": 10.0},
|
|
"revenue_growth": {"absolute": 2.0, "relative_pct": None},
|
|
"earnings_surprise": {"absolute": 2.0, "relative_pct": None},
|
|
}
|
|
DEFINITION_NOTES = {
|
|
"pe_ratio": (
|
|
"Legacy provider P/E convention versus latest close divided by "
|
|
"SEC-derived TTM diluted EPS."
|
|
),
|
|
"revenue_growth": (
|
|
"Legacy provider growth convention versus SEC-derived TTM revenue YoY."
|
|
),
|
|
"earnings_surprise": (
|
|
"Legacy provider latest surprise versus latest completed Dolt earnings "
|
|
"event with actual and estimate."
|
|
),
|
|
}
|
|
|
|
|
|
def fundamental_score(
|
|
pe_ratio: float | None,
|
|
revenue_growth: float | None,
|
|
earnings_surprise: float | None,
|
|
) -> float | None:
|
|
"""Match the production fundamental-dimension formula without persistence."""
|
|
scores: list[float] = []
|
|
if _finite(pe_ratio) and pe_ratio > 0:
|
|
scores.append(max(0.0, min(100.0, 100.0 - (pe_ratio - 15.0) * (100.0 / 30.0))))
|
|
if _finite(revenue_growth):
|
|
scores.append(max(0.0, min(100.0, 50.0 + revenue_growth * 2.5)))
|
|
if _finite(earnings_surprise):
|
|
scores.append(max(0.0, min(100.0, 50.0 + earnings_surprise * 5.0)))
|
|
return sum(scores) / len(scores) if len(scores) >= MIN_SCORE_METRICS else None
|
|
|
|
|
|
async def build_report(
|
|
db: AsyncSession,
|
|
*,
|
|
generated_at: datetime | None = None,
|
|
today: date | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Build a point-in-time parity report from one database session."""
|
|
generated_at = generated_at or datetime.now(timezone.utc)
|
|
today = today or datetime.now(ZoneInfo("America/New_York")).date()
|
|
|
|
# A report must not mix rows from before and after a concurrent import
|
|
# promotion. The scheduled job provides a fresh session, so establish the
|
|
# production snapshot before its first query and have Postgres enforce the
|
|
# no-write contract as well. SQLite tests retain their normal transaction.
|
|
if db.get_bind().dialect.name == "postgresql":
|
|
connection = await db.connection(
|
|
execution_options={"isolation_level": "REPEATABLE READ"}
|
|
)
|
|
await connection.execute(text("SET TRANSACTION READ ONLY"))
|
|
|
|
candidates = await candidate_service.build_candidates(db, today=today)
|
|
ticker_ids = [candidate.ticker_id for candidate in candidates]
|
|
legacy_by_ticker = await _legacy_values(db, ticker_ids)
|
|
source_runs = await _source_runs(db)
|
|
|
|
rows: list[dict[str, Any]] = []
|
|
for candidate in candidates:
|
|
legacy = legacy_by_ticker.get(candidate.ticker_id)
|
|
candidate_values = {
|
|
"pe_ratio": candidate.pe_ratio,
|
|
"revenue_growth": candidate.revenue_growth,
|
|
"earnings_surprise": candidate.earnings_surprise,
|
|
}
|
|
legacy_values = {
|
|
"pe_ratio": legacy.pe_ratio if legacy else None,
|
|
"revenue_growth": legacy.revenue_growth if legacy else None,
|
|
"earnings_surprise": legacy.earnings_surprise if legacy else None,
|
|
}
|
|
fields = {
|
|
key: _field_comparison(key, legacy_values[key], candidate_values[key])
|
|
for key in FIELD_KEYS
|
|
}
|
|
legacy_score = fundamental_score(**legacy_values)
|
|
candidate_score = fundamental_score(**candidate_values)
|
|
rows.append(
|
|
{
|
|
"symbol": candidate.symbol,
|
|
"cik": candidate.cik,
|
|
"legacy_fetched_at": _iso(legacy.fetched_at) if legacy else None,
|
|
"price_date": _iso(candidate.price_date),
|
|
"fields": fields,
|
|
"scores": {
|
|
"legacy_fundamental": _round(legacy_score),
|
|
"candidate_fundamental": _round(candidate_score),
|
|
"fundamental_delta": _delta(legacy_score, candidate_score),
|
|
"legacy_fundamental_rank": None,
|
|
"candidate_fundamental_rank": None,
|
|
"fundamental_rank_change": None,
|
|
},
|
|
}
|
|
)
|
|
|
|
_attach_ranks(rows, "legacy_fundamental", "legacy_fundamental_rank")
|
|
_attach_ranks(rows, "candidate_fundamental", "candidate_fundamental_rank")
|
|
for row in rows:
|
|
scores = row["scores"]
|
|
scores["fundamental_rank_change"] = _rank_change(
|
|
scores["legacy_fundamental_rank"], scores["candidate_fundamental_rank"]
|
|
)
|
|
|
|
return {
|
|
"report_version": REPORT_VERSION,
|
|
"generated_at": generated_at.isoformat(),
|
|
"as_of_date": today.isoformat(),
|
|
"approval_status": APPROVAL_STATUS,
|
|
"read_only": True,
|
|
"fundamental_score_formula": (
|
|
"Equal-weighted mean of 2+ available sub-scores: P/E = "
|
|
"clamp(100-(pe-15)*(100/30)); revenue growth = "
|
|
"clamp(50+growth*2.5); earnings surprise = "
|
|
"clamp(50+surprise*5)."
|
|
),
|
|
"source_runs": source_runs,
|
|
"definition_notes": DEFINITION_NOTES,
|
|
"materiality_notes": {
|
|
"fields": FIELD_TOLERANCES,
|
|
"fundamental_score_absolute": 5.0,
|
|
"automatic_cutover": False,
|
|
},
|
|
"summary": _summary(rows),
|
|
"rows": rows,
|
|
}
|
|
|
|
|
|
def store_report(report: dict[str, Any], report_dir: str | Path) -> dict[str, str]:
|
|
"""Atomically archive JSON/CSV artifacts and update the latest manifest."""
|
|
directory = Path(report_dir).expanduser().resolve()
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
stamp = _artifact_stamp(report["generated_at"])
|
|
json_name = f"fundamentals-parity-{stamp}.json"
|
|
csv_name = f"fundamentals-parity-{stamp}.csv"
|
|
json_path = directory / json_name
|
|
csv_path = directory / csv_name
|
|
|
|
_atomic_write(json_path, json.dumps(report, indent=2, sort_keys=True) + "\n")
|
|
_atomic_write(csv_path, report_csv(report))
|
|
manifest = {
|
|
"generated_at": report["generated_at"],
|
|
"json_file": json_name,
|
|
"csv_file": csv_name,
|
|
}
|
|
_atomic_write(
|
|
directory / "latest.json",
|
|
json.dumps(manifest, indent=2, sort_keys=True) + "\n",
|
|
)
|
|
return {
|
|
"json": str(json_path),
|
|
"csv": str(csv_path),
|
|
"manifest": str(directory / "latest.json"),
|
|
}
|
|
|
|
|
|
async def generate_and_store(
|
|
db: AsyncSession,
|
|
report_dir: str | Path,
|
|
*,
|
|
generated_at: datetime | None = None,
|
|
today: date | None = None,
|
|
) -> tuple[dict[str, Any], dict[str, str]]:
|
|
report = await build_report(db, generated_at=generated_at, today=today)
|
|
return report, store_report(report, report_dir)
|
|
|
|
|
|
def load_latest(report_dir: str | Path) -> dict[str, Any] | None:
|
|
manifest = _load_manifest(report_dir)
|
|
if manifest is None:
|
|
return None
|
|
try:
|
|
path = _manifest_artifact(report_dir, manifest, "json_file")
|
|
loaded = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError, TypeError, ValueError):
|
|
return None
|
|
return loaded if isinstance(loaded, dict) else None
|
|
|
|
|
|
def load_latest_csv(report_dir: str | Path) -> tuple[str, str] | None:
|
|
return _load_latest_text_artifact(report_dir, "csv_file")
|
|
|
|
|
|
def load_latest_json(report_dir: str | Path) -> tuple[str, str] | None:
|
|
return _load_latest_text_artifact(report_dir, "json_file")
|
|
|
|
|
|
def _load_latest_text_artifact(
|
|
report_dir: str | Path, manifest_key: str
|
|
) -> tuple[str, str] | None:
|
|
manifest = _load_manifest(report_dir)
|
|
if manifest is None:
|
|
return None
|
|
try:
|
|
path = _manifest_artifact(report_dir, manifest, manifest_key)
|
|
return path.name, path.read_text(encoding="utf-8")
|
|
except (OSError, TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def report_csv(report: dict[str, Any]) -> str:
|
|
output = io.StringIO(newline="")
|
|
columns = [
|
|
"symbol",
|
|
"cik",
|
|
"legacy_fetched_at",
|
|
"price_date",
|
|
*(
|
|
f"{field}_{suffix}"
|
|
for field in FIELD_KEYS
|
|
for suffix in ("legacy", "candidate", "absolute_delta", "relative_delta_pct", "material")
|
|
),
|
|
"legacy_fundamental",
|
|
"candidate_fundamental",
|
|
"fundamental_delta",
|
|
"legacy_fundamental_rank",
|
|
"candidate_fundamental_rank",
|
|
"fundamental_rank_change",
|
|
]
|
|
writer = csv.DictWriter(output, fieldnames=columns)
|
|
writer.writeheader()
|
|
for row in report.get("rows", []):
|
|
flat = {
|
|
"symbol": row["symbol"],
|
|
"cik": row.get("cik"),
|
|
"legacy_fetched_at": row.get("legacy_fetched_at"),
|
|
"price_date": row.get("price_date"),
|
|
**row["scores"],
|
|
}
|
|
for field in FIELD_KEYS:
|
|
comparison = row["fields"][field]
|
|
for suffix in (
|
|
"legacy",
|
|
"candidate",
|
|
"absolute_delta",
|
|
"relative_delta_pct",
|
|
"material",
|
|
):
|
|
flat[f"{field}_{suffix}"] = comparison.get(suffix)
|
|
writer.writerow(flat)
|
|
return output.getvalue()
|
|
|
|
|
|
async def _legacy_values(
|
|
db: AsyncSession, ticker_ids: list[int]
|
|
) -> dict[int, FundamentalData]:
|
|
if not ticker_ids:
|
|
return {}
|
|
rows = (
|
|
await db.execute(
|
|
select(FundamentalData).where(FundamentalData.ticker_id.in_(ticker_ids))
|
|
)
|
|
).scalars()
|
|
return {row.ticker_id: row for row in rows}
|
|
|
|
|
|
async def _source_runs(db: AsyncSession) -> dict[str, dict[str, Any] | None]:
|
|
sources = ("sec_facts", "dolt_earnings")
|
|
rows = (
|
|
await db.execute(
|
|
select(DataImportRun)
|
|
.where(
|
|
DataImportRun.source.in_(sources),
|
|
DataImportRun.status.in_(("promoted", "no_op")),
|
|
)
|
|
.order_by(DataImportRun.id.desc())
|
|
)
|
|
).scalars()
|
|
latest: dict[str, dict[str, Any] | None] = {source: None for source in sources}
|
|
for row in rows:
|
|
if latest[row.source] is None:
|
|
latest[row.source] = {
|
|
"run_id": row.id,
|
|
"status": row.status,
|
|
"revision": row.revision,
|
|
"source_max_date": _iso(row.source_max_date),
|
|
"completed_at": _iso(row.completed_at),
|
|
}
|
|
return latest
|
|
|
|
|
|
def _field_comparison(
|
|
key: str, legacy: float | None, candidate: float | None
|
|
) -> dict[str, Any]:
|
|
legacy = float(legacy) if _finite(legacy) else None
|
|
candidate = float(candidate) if _finite(candidate) else None
|
|
absolute = _delta(legacy, candidate)
|
|
relative = (
|
|
None
|
|
if absolute is None or legacy in (None, 0)
|
|
else round(absolute / abs(legacy) * 100.0, 4)
|
|
)
|
|
tolerance = FIELD_TOLERANCES[key]
|
|
material = False
|
|
if absolute is not None:
|
|
material = abs(absolute) > tolerance["absolute"]
|
|
relative_limit = tolerance["relative_pct"]
|
|
if relative_limit is not None:
|
|
material = material and relative is not None and abs(relative) > relative_limit
|
|
return {
|
|
"legacy": _round(legacy),
|
|
"candidate": _round(candidate),
|
|
"absolute_delta": absolute,
|
|
"relative_delta_pct": relative,
|
|
"material": material,
|
|
"definition_changed": True,
|
|
}
|
|
|
|
|
|
def _attach_ranks(rows: list[dict[str, Any]], value_key: str, rank_key: str) -> None:
|
|
values = [
|
|
row["scores"][value_key]
|
|
for row in rows
|
|
if _finite(row["scores"][value_key])
|
|
]
|
|
for row in rows:
|
|
value = row["scores"][value_key]
|
|
row["scores"][rank_key] = (
|
|
1 + sum(other > value for other in values) if _finite(value) else None
|
|
)
|
|
|
|
|
|
def _summary(rows: list[dict[str, Any]]) -> dict[str, Any]:
|
|
field_stats = {}
|
|
for key in FIELD_KEYS:
|
|
comparisons = [row["fields"][key] for row in rows]
|
|
deltas = [
|
|
abs(item["absolute_delta"])
|
|
for item in comparisons
|
|
if item["absolute_delta"] is not None
|
|
]
|
|
field_stats[key] = {
|
|
"legacy_available": sum(item["legacy"] is not None for item in comparisons),
|
|
"candidate_available": sum(
|
|
item["candidate"] is not None for item in comparisons
|
|
),
|
|
"both_available": len(deltas),
|
|
"material_differences": sum(item["material"] for item in comparisons),
|
|
"median_absolute_delta": _round(statistics.median(deltas) if deltas else None),
|
|
"p95_absolute_delta": _round(_percentile(deltas, 0.95)),
|
|
"max_absolute_delta": _round(max(deltas) if deltas else None),
|
|
}
|
|
|
|
fundamental_deltas = _score_deltas(rows, "fundamental_delta")
|
|
changed_rows = sorted(
|
|
(
|
|
{
|
|
"symbol": row["symbol"],
|
|
"fundamental_delta": row["scores"]["fundamental_delta"],
|
|
"fundamental_rank_change": row["scores"]["fundamental_rank_change"],
|
|
}
|
|
for row in rows
|
|
if row["scores"]["fundamental_delta"] is not None
|
|
),
|
|
key=lambda item: (
|
|
abs(item["fundamental_delta"] or 0),
|
|
),
|
|
reverse=True,
|
|
)[:20]
|
|
return {
|
|
"universe_count": len(rows),
|
|
"legacy_fundamental_score_available": _count_score(
|
|
rows, "legacy_fundamental"
|
|
),
|
|
"candidate_fundamental_score_available": _count_score(
|
|
rows, "candidate_fundamental"
|
|
),
|
|
"fundamental_scores_compared": len(fundamental_deltas),
|
|
"fundamental_score_material_changes": sum(
|
|
abs(delta) > 5.0 for delta in fundamental_deltas
|
|
),
|
|
"fundamental_rank_changes": _rank_change_count(
|
|
rows, "fundamental_rank_change"
|
|
),
|
|
"field_stats": field_stats,
|
|
"largest_changes": changed_rows,
|
|
}
|
|
|
|
|
|
def _score_deltas(rows: Iterable[dict[str, Any]], key: str) -> list[float]:
|
|
return [
|
|
row["scores"][key]
|
|
for row in rows
|
|
if row["scores"][key] is not None
|
|
]
|
|
|
|
|
|
def _count_score(rows: Iterable[dict[str, Any]], key: str) -> int:
|
|
return sum(row["scores"][key] is not None for row in rows)
|
|
|
|
|
|
def _rank_change_count(rows: Iterable[dict[str, Any]], key: str) -> int:
|
|
return sum(
|
|
row["scores"][key] not in (None, 0)
|
|
for row in rows
|
|
)
|
|
|
|
|
|
def _rank_change(legacy: int | None, candidate: int | None) -> int | None:
|
|
# Positive means the candidate improved its rank.
|
|
return legacy - candidate if legacy is not None and candidate is not None else None
|
|
|
|
|
|
def _delta(legacy: float | None, candidate: float | None) -> float | None:
|
|
if not _finite(legacy) or not _finite(candidate):
|
|
return None
|
|
return round(candidate - legacy, 4)
|
|
|
|
|
|
def _round(value: float | None, digits: int = 4) -> float | None:
|
|
return round(float(value), digits) if _finite(value) else None
|
|
|
|
|
|
def _percentile(values: list[float], quantile: float) -> float | None:
|
|
if not values:
|
|
return None
|
|
ordered = sorted(values)
|
|
index = max(0, math.ceil(quantile * len(ordered)) - 1)
|
|
return ordered[index]
|
|
|
|
|
|
def _finite(value: Any) -> bool:
|
|
return (
|
|
isinstance(value, (int, float))
|
|
and not isinstance(value, bool)
|
|
and math.isfinite(value)
|
|
)
|
|
|
|
|
|
def _iso(value: Any) -> str | None:
|
|
return value.isoformat() if value is not None else None
|
|
|
|
|
|
def _artifact_stamp(raw: str) -> str:
|
|
parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
|
|
return parsed.astimezone(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
|
|
|
|
|
|
def _atomic_write(path: Path, content: str) -> None:
|
|
temp = path.with_name(f".{path.name}.{os.getpid()}.tmp")
|
|
temp.write_text(content, encoding="utf-8", newline="")
|
|
os.replace(temp, path)
|
|
|
|
|
|
def _load_manifest(report_dir: str | Path) -> dict[str, Any] | None:
|
|
path = Path(report_dir).expanduser().resolve() / "latest.json"
|
|
try:
|
|
loaded = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError, TypeError, ValueError):
|
|
return None
|
|
return loaded if isinstance(loaded, dict) else None
|
|
|
|
|
|
def _manifest_artifact(
|
|
report_dir: str | Path, manifest: dict[str, Any], key: str
|
|
) -> Path:
|
|
directory = Path(report_dir).expanduser().resolve()
|
|
name = Path(str(manifest.get(key, ""))).name
|
|
if not name:
|
|
raise ValueError(f"Latest parity manifest has no {key}")
|
|
return directory / name
|