Spaces:
Sleeping
Sleeping
File size: 5,085 Bytes
2f5da7d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 | """The data-foundation acceptance criteria, as queries rather than assertions.
Spec: docs/superpowers/specs/2026-09-01-data-foundation-design.md
Three criteria deliberately permit a shortfall provided it is enumerated. A
filer that does not tag a concept is a fact about the filer, and recording it
is the correct outcome; reporting a quietly lower number is not. So every
criterion that can fall short prints the exact ticker, period and concept.
Usage::
python -m scripts.data_foundation_report
"""
from __future__ import annotations
import json
import sqlite3
import sys
DB_PATH = "data/metrics.db"
NEW_STATEMENT_LINES = (
"deferred_revenue", "research_development", "sga", "goodwill",
"intangibles", "cash_and_st_investments", "debt_current",
"operating_lease_liabilities",
)
CORE_STATEMENT_LINES = (
"operating_cash_flow", "net_income", "total_assets",
"receivables", "payables", "cogs",
)
GUIDANCE_FLOOR = 40
def _rows() -> list[dict]:
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
try:
return [dict(r) for r in conn.execute("SELECT * FROM metrics")]
finally:
conn.close()
def _segment_tickers() -> set[str]:
conn = sqlite3.connect(DB_PATH)
try:
return {r[0] for r in conn.execute("SELECT DISTINCT ticker FROM segments")}
except sqlite3.OperationalError:
return set()
finally:
conn.close()
def build_report() -> list[dict]:
rows = _rows()
total = len(rows)
out: list[dict] = []
legacy = [f"{r['ticker']} {r['period']}" for r in rows
if r.get("data_quality_status") == "LEGACY_UNVERIFIED"]
out.append({
"criterion": "no_legacy_unverified_rows", "target": "0",
"actual": str(len(legacy)), "passed": not legacy, "detail": legacy,
})
no_url = [f"{r['ticker']} {r['period']}" for r in rows if not r.get("source_url")]
out.append({
"criterion": "every_row_carries_source_url", "target": "0 missing",
"actual": str(len(no_url)), "passed": not no_url, "detail": no_url,
})
parsed = [r for r in rows if r.get("guidance_status") == "PARSED"]
silent = [f"{r['ticker']} {r['period']}" for r in rows
if r.get("guidance_status") == "NO_GUIDANCE_ISSUED"]
errored = [f"{r['ticker']} {r['period']}" for r in rows
if r.get("guidance_status") == "PARSE_ERROR"]
unquoted = [f"{r['ticker']} {r['period']}" for r in parsed
if not r.get("guidance_source_quote")]
out.append({
"criterion": "guidance_parsed_above_floor", "target": f">= {GUIDANCE_FLOOR}",
"actual": str(len(parsed)), "passed": len(parsed) >= GUIDANCE_FLOOR,
"detail": [f"no guidance issued: {len(silent)}", f"parse errors: {len(errored)}"]
+ errored,
})
out.append({
"criterion": "every_parsed_guidance_has_a_quote", "target": "0 unquoted",
"actual": str(len(unquoted)), "passed": not unquoted, "detail": unquoted,
})
core_gaps = [
f"{r['ticker']} {r['period']} {line}"
for r in rows for line in CORE_STATEMENT_LINES if r.get(line) is None
]
out.append({
"criterion": "core_statement_lines_populated", "target": f"{total} rows each",
"actual": f"{len(core_gaps)} gaps", "passed": not core_gaps, "detail": core_gaps,
})
new_gaps = [
f"{r['ticker']} {r['period']} {line}"
for r in rows for line in NEW_STATEMENT_LINES if r.get(line) is None
]
out.append({
"criterion": "new_statement_lines_populated",
"target": "populated wherever the filer tags the concept",
"actual": f"{len(new_gaps)} absences", "passed": True, "detail": new_gaps,
})
fallback = []
for r in rows:
try:
warnings = json.loads(r.get("quality_warnings") or "[]")
except (TypeError, ValueError):
warnings = []
for w in warnings:
if "fallback_non_sec" in w:
fallback.append(f"{r['ticker']} {r['period']} {w}")
out.append({
"criterion": "no_fallback_non_sec", "target": "0",
"actual": str(len(fallback)), "passed": not fallback, "detail": fallback,
})
covered = {r["ticker"] for r in rows}
with_segments = _segment_tickers()
missing = sorted(covered - with_segments)
out.append({
"criterion": "segments_on_every_covered_ticker",
"target": f"{len(covered)} tickers",
"actual": str(len(with_segments)), "passed": not missing, "detail": missing,
})
return out
def main() -> None:
report = build_report()
for r in report:
mark = "PASS" if r["passed"] else "FAIL"
print(f"[{mark}] {r['criterion']}: {r['actual']} (target {r['target']})")
for line in r["detail"][:40]:
print(f" {line}")
if len(r["detail"]) > 40:
print(f" ... and {len(r['detail']) - 40} more")
failed = [r for r in report if not r["passed"]]
sys.exit(1 if failed else 0)
if __name__ == "__main__":
main()
|