Spaces:
Sleeping
Sleeping
File size: 5,929 Bytes
231e504 | 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 | #!/usr/bin/env python3
"""Validate internal consistency of financials.json and sectors.json."""
import json
import sys
from pathlib import Path
DATA_DIR = Path(__file__).parent.parent / "server" / "data"
def main():
errors = []
with open(DATA_DIR / "financials.json") as f:
data = json.load(f)
with open(DATA_DIR / "sectors.json") as f:
sectors = json.load(f)
required_tickers = {"AAPL", "MSFT", "GOOGL", "META", "NVDA", "TSLA", "F", "GM", "JPM", "BAC", "AMZN", "WMT"}
required_years = {"2019", "2020", "2021", "2022", "2023"}
# Check all tickers present
missing_tickers = required_tickers - set(data.keys())
if missing_tickers:
errors.append(f"Missing tickers: {missing_tickers}")
for ticker in data:
# Check all years present
missing_years = required_years - set(data[ticker].keys())
if missing_years:
errors.append(f"{ticker}: missing years {missing_years}")
for year in data[ticker]:
e = data[ticker][year]
inc = e["income_statement"]
bs = e["balance_sheet"]
cf = e["cash_flow"]
# --- Core invariants ---
if inc["gross_profit"] != inc["revenue"] - inc["cogs"]:
errors.append(f"{ticker}/{year}: gross_profit != revenue - cogs")
if bs["total_equity"] != bs["total_assets"] - bs["total_liabilities"]:
errors.append(f"{ticker}/{year}: total_equity != total_assets - total_liabilities")
if cf["fcf"] != cf["operating_cf"] - cf["capex"]:
errors.append(f"{ticker}/{year}: fcf != operating_cf - capex")
if inc["operating_income"] >= inc["gross_profit"]:
errors.append(f"{ticker}/{year}: operating_income >= gross_profit")
# --- Required fields ---
for field in ["revenue", "cogs", "gross_profit", "operating_income", "net_income", "eps"]:
if field not in inc:
errors.append(f"{ticker}/{year}: missing income_statement.{field}")
for field in ["total_assets", "total_liabilities", "total_equity", "cash", "total_debt"]:
if field not in bs:
errors.append(f"{ticker}/{year}: missing balance_sheet.{field}")
for field in ["operating_cf", "investing_cf", "financing_cf", "fcf", "capex"]:
if field not in cf:
errors.append(f"{ticker}/{year}: missing cash_flow.{field}")
# Price
if "price" not in e:
errors.append(f"{ticker}/{year}: missing price")
else:
for field in ["open", "close", "high", "low", "avg_price"]:
if field not in e["price"]:
errors.append(f"{ticker}/{year}: missing price.{field}")
if "shares_outstanding" not in e:
errors.append(f"{ticker}/{year}: missing shares_outstanding")
# --- Ratios block ---
if "ratios" not in e:
errors.append(f"{ticker}/{year}: missing ratios block")
else:
r = e["ratios"]
for field in ["pe_ratio", "pb_ratio", "ev_ebitda", "roe", "roa",
"debt_equity", "current_ratio", "gross_margin", "net_margin", "fcf_margin"]:
if field not in r:
errors.append(f"{ticker}/{year}: missing ratios.{field}")
# Margin invariants (within 0.001)
if r.get("gross_margin") is not None and inc["revenue"] > 0:
expected = inc["gross_profit"] / inc["revenue"]
if abs(r["gross_margin"] - expected) > 0.001:
errors.append(f"{ticker}/{year}: gross_margin {r['gross_margin']} != {expected:.4f}")
if r.get("net_margin") is not None and inc["revenue"] > 0:
expected = inc["net_income"] / inc["revenue"]
if abs(r["net_margin"] - expected) > 0.001:
errors.append(f"{ticker}/{year}: net_margin {r['net_margin']} != {expected:.4f}")
# --- Sectors ---
required_sectors = {"technology", "automotive", "banking", "retail"}
actual_sectors = {k for k in sectors if k != "_ticker_sectors"}
missing_sectors = required_sectors - actual_sectors
if missing_sectors:
errors.append(f"Missing sectors: {missing_sectors}")
for sector in actual_sectors:
missing_syears = required_years - set(sectors[sector].keys())
if missing_syears:
errors.append(f"Sector {sector}: missing years {missing_syears}")
# Ticker-sector mapping
ticker_sectors = sectors.get("_ticker_sectors", {})
for ticker in required_tickers:
if ticker not in ticker_sectors:
errors.append(f"Missing sector mapping for {ticker}")
# Task-specific: TSLA neg FCF in 2019 and 2020
for y in ["2019", "2020"]:
if data["TSLA"][y]["cash_flow"]["fcf"] >= 0:
errors.append(f"TSLA/{y}: FCF should be negative for task3")
# META revenue decline 2021->2022
if data["META"]["2022"]["income_statement"]["revenue"] >= data["META"]["2021"]["income_statement"]["revenue"]:
errors.append("META: revenue should decline from 2021 to 2022")
if errors:
print(f"FAILED -- {len(errors)} error(s):")
for e in errors:
print(f" {e}")
sys.exit(1)
else:
tickers = sorted(data.keys())
print(f"PASSED -- {len(tickers)} tickers, {len(required_years)} years each")
print(f" Tickers: {', '.join(tickers)}")
print(f" Sectors: {', '.join(sorted(actual_sectors))}")
print(f" Ratios: all 10 fields present per ticker/year")
print(f" Invariants: gross_profit, equity, fcf, gross_margin, net_margin")
sys.exit(0)
if __name__ == "__main__":
main()
|