mirror of
https://github.com/alirezarezvani/claude-skills.git
synced 2026-10-07 02:58:13 +00:00
1663 lines
72 KiB
Python
1663 lines
72 KiB
Python
#!/usr/bin/env python3
|
|
"""verify_data.py -- data-intake verification gate for the stock-analysis skill.
|
|
|
|
Purpose
|
|
-------
|
|
Blinded adversarial review of this skill's real reports found the dominant
|
|
weakness is not analysis quality but DATA INTEGRITY: figures that could not be
|
|
traced to a source, a peer's NPA stated wrong, and reports that silently omitted
|
|
a quarter which was already public on the analysis date. Those are intake
|
|
defects, not reasoning defects -- and an intake defect contaminates every ratio,
|
|
every peer comparison and the final verdict, invisibly and confidently.
|
|
|
|
This script runs the data the agent has *gathered* through a gate BEFORE the
|
|
analysis touches it, so the defect is caught at intake instead of surfacing in
|
|
the verdict. It reads a JSON description of the gathered datapoints, applies a
|
|
fixed set of provenance / consistency / recency rules, and prints findings with
|
|
a PASS / PASS-WITH-WARNINGS / FAIL headline. It exits non-zero on any
|
|
error-severity finding so it can gate a pipeline.
|
|
|
|
It never fetches anything and never invents anything. It only checks the
|
|
internal consistency, completeness and freshness of what it was given. A clean
|
|
PASS here does not mean the numbers are *right* -- it means they are traceable,
|
|
mutually consistent and current enough to analyse. Those are the preconditions
|
|
the review found were being skipped.
|
|
|
|
Standard library only. Python 3.8+. No network, no third-party packages.
|
|
|
|
--------------------------------------------------------------------------------
|
|
INPUT SCHEMA
|
|
--------------------------------------------------------------------------------
|
|
A JSON object. Top-level fields:
|
|
|
|
company str -- issuer name (free text).
|
|
ticker str -- exchange:symbol, e.g. "NSE:MERIDBANK".
|
|
as_of str -- ISO date the data was gathered, "2026-07-23".
|
|
Staleness is measured from THIS, never from
|
|
the wall clock -- the gate must be reproducible.
|
|
reporting_basis str -- "consolidated" or "standalone"; the basis the
|
|
analysis will be run on.
|
|
currency str -- file-level currency, e.g. "INR", "USD".
|
|
units str -- file-level scale, e.g. "crore", "million".
|
|
fiscal_year_end str -- optional. Month the fiscal year ends, e.g.
|
|
"March" or 3. Defaults to March (Indian
|
|
convention). Governs how "FY26"/"Q1 FY27"
|
|
labels are converted to period-end dates.
|
|
latest_reported_period obj -- {"period": "Q1 FY27", "published": "2026-07-11"}
|
|
The newest period the company has ACTUALLY
|
|
reported (from an exchange/filing check), which
|
|
may be newer than anything gathered below.
|
|
datapoints list -- the gathered figures (see below).
|
|
not_available list -- optional. Metric names the agent could not
|
|
source and has explicitly left blank. Declaring
|
|
these is GOOD discipline and is reported as
|
|
coverage context, never as a defect.
|
|
|
|
Each entry in "datapoints" is an object:
|
|
|
|
metric str -- REQUIRED. Canonical metric name, e.g. "revenue",
|
|
"net_profit", "gross_npa_pct". Grouping is by this name.
|
|
value num -- the figure. May be null or "not available" to mark the metric
|
|
as explicitly unsourced (counts toward coverage, not defects).
|
|
period str -- REQUIRED for a real figure. "FY26", "Q1 FY27", "CY2025",
|
|
"TTM to Sep-2025", "year ended Dec-2025".
|
|
basis str -- "consolidated" or "standalone".
|
|
unit str -- e.g. "INR crore", "USD million", "%", "x", "days".
|
|
source str -- REQUIRED for a real figure. Where it came from, with enough
|
|
detail to re-find it: "FY26 annual report, p.142".
|
|
source_tier int -- optional 1-4. 1 primary filing (annual report/10-K/exchange
|
|
results/DRHP), 2 company-published secondary (presentation,
|
|
concall, release), 3 regulator/third-party primary (rating
|
|
rationale, MCA/ROC, SEBI, shareholding pattern), 4 aggregator
|
|
(screener/tickertape/yahoo/moneycontrol...). If omitted it is
|
|
INFERRED from the `source` text. Tier 4 is a navigation aid,
|
|
never a source of record: a HEADLINE metric (revenue, PAT,
|
|
EBITDA, debt, equity, operating cash flow) that is Tier-4-only
|
|
is an ERROR. Current price/market cap is the one sanctioned
|
|
exception and is exempt.
|
|
entity str -- optional. Whose figure this is, when it is not the subject
|
|
company -- a peer's, for a benchmark. Defaults to `company`.
|
|
Lets the peer's-NPA-stated-wrong case be period-aligned and
|
|
cross-checked in its own right.
|
|
note str -- optional. A free-text acknowledgement that legitimises an
|
|
otherwise-flagged deviation (e.g. a deliberately standalone
|
|
figure, or a figure in a different unit). Its PRESENCE
|
|
downgrades basis/unit mismatches from silent to disclosed.
|
|
alt list -- optional. Cross-source corroborating reads of the SAME figure:
|
|
[{"value": 59480, "source": "screener.in 2026-07-21"}].
|
|
The gate compares each against `value`.
|
|
|
|
The template printed by --template is annotated with these rules. Comment lines
|
|
(whole lines whose first non-space characters are `//`) are stripped by the
|
|
loader, so a filled-in template loads directly; keep comments on their own line.
|
|
|
|
--------------------------------------------------------------------------------
|
|
CHECKS
|
|
--------------------------------------------------------------------------------
|
|
1 PROVENANCE source/period missing = error; unit/basis missing = warn.
|
|
2 CROSS-SOURCE `alt` vs `value` divergence: >2% warn, >10% error.
|
|
3 BASIS MIXING consolidated and standalone both present, unlabelled = error.
|
|
4 UNIT/CURRENCY unit/currency differs from the file declaration = warn;
|
|
a clean 10x/100x gap between comparable figures = error.
|
|
5 PERIOD ALIGN comparing FY vs CY vs TTM, or misaligned year-ends = warn.
|
|
6 RECENCY a newer reported period exists but is not incorporated, or
|
|
as_of is >~4 months past the newest period-end = warn.
|
|
7 COVERAGE count of metrics explicitly marked not-available = info.
|
|
8 SOURCE TIER a HEADLINE metric sourced only from a Tier-4 aggregator =
|
|
error; a non-headline aggregator-only figure = warn; a
|
|
headline figure whose source cannot be confirmed as a filing
|
|
= warn; plus a % primary-sourced coverage line = info.
|
|
|
|
--------------------------------------------------------------------------------
|
|
USAGE
|
|
--------------------------------------------------------------------------------
|
|
python verify_data.py intake.json # readable report
|
|
python verify_data.py intake.json --json # machine-readable findings
|
|
python verify_data.py --template # annotated blank intake
|
|
python verify_data.py --example # run the built-in example
|
|
python verify_data.py --example --json
|
|
|
|
Exit code: 0 if no error-severity finding (PASS or PASS-WITH-WARNINGS),
|
|
1 if any error-severity finding (FAIL), 2 on unusable input (bad path / bad JSON).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from datetime import date, timedelta
|
|
from typing import Any, Dict, List, Optional, Sequence, Tuple
|
|
|
|
SCRIPT = "verify_data.py"
|
|
|
|
SEVERITY_ERROR = "error"
|
|
SEVERITY_WARN = "warn"
|
|
SEVERITY_INFO = "info"
|
|
# Sort order for grouping the report: errors first, then warnings, then info.
|
|
_SEVERITY_RANK = {SEVERITY_ERROR: 0, SEVERITY_WARN: 1, SEVERITY_INFO: 2}
|
|
|
|
# Divergence thresholds for cross-source disagreement, in percent. A gap of a
|
|
# few percent is usually a definitional difference (other income in/out of
|
|
# EBITDA, leases, average vs closing balances). A gap above ~10% almost always
|
|
# means a different basis, period or unit -- i.e. the two "sources" are not
|
|
# measuring the same thing, and trusting either blindly is how a wrong figure
|
|
# (the peer-NPA case) reaches the report. So >2% is worth a human glance (warn)
|
|
# and >10% must block until reconciled (error).
|
|
DIVERGENCE_WARN_PCT = 2.0
|
|
DIVERGENCE_ERROR_PCT = 10.0
|
|
|
|
# How close a ratio between two "comparable" figures must be to a power of ten
|
|
# before it is treated as a unit/scale error rather than a real difference.
|
|
# 1 crore = 10 million, so the crore-vs-million slip lands almost exactly on 10x.
|
|
MAGNITUDE_TOLERANCE = 0.03 # +/-3% around 10x, 100x, 1000x.
|
|
MAGNITUDE_FACTORS = (10.0, 100.0, 1000.0)
|
|
|
|
# Recency thresholds, in days, measured against as_of (never the wall clock).
|
|
# ~20 days separates "same reporting period, rounding" from "a genuinely later
|
|
# period exists". ~4 months (123 days) past a period-end means at least one
|
|
# quarterly result has almost certainly been published since and was not checked.
|
|
RECENCY_NEWER_PERIOD_DAYS = 20
|
|
RECENCY_STALE_DAYS = 123
|
|
|
|
_MONTHS = {
|
|
"jan": 1, "january": 1, "feb": 2, "february": 2, "mar": 3, "march": 3,
|
|
"apr": 4, "april": 4, "may": 5, "jun": 6, "june": 6, "jul": 7, "july": 7,
|
|
"aug": 8, "august": 8, "sep": 9, "sept": 9, "september": 9, "oct": 10,
|
|
"october": 10, "nov": 11, "november": 11, "dec": 12, "december": 12,
|
|
}
|
|
|
|
# Currency spellings -> canonical code. `$` is deliberately ambiguous across
|
|
# USD/SGD/HKD/AUD/CAD; it is mapped to USD only so a `$` figure inside an INR
|
|
# file still trips the currency-mismatch check rather than passing silently.
|
|
_CURRENCY_ALIASES = {
|
|
"inr": "INR", "rs": "INR", "rs.": "INR", "₹": "INR", "rupee": "INR",
|
|
"rupees": "INR", "usd": "USD", "us$": "USD", "$": "USD", "dollar": "USD",
|
|
"eur": "EUR", "€": "EUR", "euro": "EUR", "gbp": "GBP", "£": "GBP",
|
|
"jpy": "JPY", "¥": "JPY", "yen": "JPY", "cny": "CNY", "rmb": "CNY",
|
|
"sgd": "SGD", "hkd": "HKD", "aud": "AUD", "cad": "CAD",
|
|
}
|
|
|
|
# Scale spellings -> canonical scale name.
|
|
_SCALE_ALIASES = {
|
|
"cr": "crore", "cr.": "crore", "crore": "crore", "crores": "crore",
|
|
"lakh": "lakh", "lakhs": "lakh", "lac": "lakh", "lacs": "lakh",
|
|
"mn": "million", "mm": "million", "mio": "million", "million": "million",
|
|
"millions": "million", "bn": "billion", "billion": "billion",
|
|
"billions": "billion", "k": "thousand", "thousand": "thousand",
|
|
"thousands": "thousand", "'000": "thousand", "000s": "thousand",
|
|
"trillion": "trillion", "tn": "trillion",
|
|
}
|
|
|
|
# Unit tokens that describe a ratio/percentage/count rather than a monetary
|
|
# amount. These are never compared against the file's currency/units, and never
|
|
# enter the 10x magnitude test (a 0.9% vs 9% pair must not read as a unit error).
|
|
_NON_MONETARY_TOKENS = {
|
|
"%", "percent", "pct", "bps", "bp", "x", "times", "ratio", "days", "day",
|
|
"years", "months", "count", "number", "no", "units", "per", "share",
|
|
"shares", "eps", "dps", "bv",
|
|
}
|
|
|
|
# Strings that, as a datapoint value, mean "the agent looked and it is not
|
|
# available" -- deliberate, disclosed gaps. These are coverage, not defects.
|
|
_UNAVAILABLE_RE = re.compile(
|
|
r"^\s*(n/?a|not\s+available|not\s+disclosed|not\s+sourced|unavailable|"
|
|
r"nd|--|-)\s*$",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
# ---- Source-tier vocabulary (Check 8) ------------------------------------- #
|
|
# The skill is document-first: figures must trace to the company's own filings,
|
|
# and aggregators (screener/tickertape/yahoo...) are navigation aids, never a
|
|
# source of record. These patterns infer a tier (1-4) from a free-text `source`
|
|
# when the datapoint does not state `source_tier` explicitly. Priority is
|
|
# highest-tier-wins: a source naming both a filing and an aggregator ("FY26 AR,
|
|
# cross-checked screener") is Tier 1, because it means the figure was traced to
|
|
# the filing.
|
|
_TIER1_SOURCE_RE = re.compile(
|
|
r"annual report|integrated report|10-?k|10-?q|20-?f|8-?k|6-?k|"
|
|
r"quarterly results|results filing|results intimation|audited|"
|
|
r"financial statements?|balance sheet|cash ?flow statement|"
|
|
r"statement of profit|profit (?:&|and) loss|prospectus|drhp|rhp|"
|
|
r"\bs-?1\b|\bf-?1\b|red herring|offer document|exchange filing|"
|
|
r"bse filing|nse filing|regulation ?30|\bcaro\b|financial report",
|
|
re.IGNORECASE,
|
|
)
|
|
_TIER2_SOURCE_RE = re.compile(
|
|
r"investor presentation|investor deck|earnings call|con-?call|"
|
|
r"earnings transcript|\btranscript\b|earnings release|press release|"
|
|
r"fact ?sheet|investor update|analyst meet|management commentary",
|
|
re.IGNORECASE,
|
|
)
|
|
_TIER3_SOURCE_RE = re.compile(
|
|
r"rating rationale|credit rating|crisil|icra|care ratings|india ratings|"
|
|
r"acuit|brickwork|\bmca\b|\broc\b|registrar of companies|\bsebi\b|"
|
|
r"shareholding pattern|bulk deal|block deal|\brbi\b|irdai|edgar|"
|
|
r"pillar ?3|form ?4\b|13-?[dfg]\b|def ?14a",
|
|
re.IGNORECASE,
|
|
)
|
|
_TIER4_SOURCE_RE = re.compile(
|
|
r"screener|tikr|tijori|trendlyne|tickertape|moneycontrol|"
|
|
r"yahoo\s*finance|google\s*finance|marketsmojo|investing\.com|stockanalysis|"
|
|
r"wsj\.com|morningstar|simplywall|value\s*research|valueresearch|"
|
|
r"finology|equitymaster|5paisa|\bgroww\b|angelone|zerodha|wikipedia|"
|
|
r"\baggregator\b|capitaline|smart-?investing|marketscreener",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
# Metrics whose reliability drives the whole verdict: an aggregator-only read of
|
|
# any of these is an error, not a warning.
|
|
_HEADLINE_METRIC_RE = re.compile(
|
|
r"revenue|turnover|net\s*sales|\bsales\b|net\s*profit|\bpat\b|"
|
|
r"profit after tax|net income|ebitda|operating profit|"
|
|
r"total\s*debt|borrowing|net\s*debt|\bequity\b|net\s*worth|"
|
|
r"shareholders?\s*funds|operating cash|cash from operations|\bcfo\b|\bocf\b",
|
|
re.IGNORECASE,
|
|
)
|
|
# Market data legitimately sourced from an exchange/finance site -- the one
|
|
# sanctioned exception to the document-first rule. Exempt from the tier check.
|
|
_MARKET_DATA_METRIC_RE = re.compile(
|
|
r"price|market\s*cap|mcap|market capitalis|52-?week|enterprise value|\bev\b",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Finding model
|
|
# --------------------------------------------------------------------------- #
|
|
@dataclass
|
|
class Finding:
|
|
"""A single verification result.
|
|
|
|
rule_id -- stable, greppable identifier for the rule that fired.
|
|
severity -- error | warn | info.
|
|
message -- human-readable sentence that NAMES the offending datapoint(s).
|
|
metric / period / entity -- structured anchors, echoed in --json so a
|
|
pipeline can route the finding without parsing
|
|
the message.
|
|
"""
|
|
|
|
rule_id: str
|
|
severity: str
|
|
message: str
|
|
metric: Optional[str] = None
|
|
period: Optional[str] = None
|
|
entity: Optional[str] = None
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Return the finding as a plain dict for the --json output."""
|
|
out: Dict[str, Any] = {
|
|
"rule_id": self.rule_id,
|
|
"severity": self.severity,
|
|
"message": self.message,
|
|
}
|
|
if self.metric is not None:
|
|
out["metric"] = self.metric
|
|
if self.period is not None:
|
|
out["period"] = self.period
|
|
if self.entity is not None:
|
|
out["entity"] = self.entity
|
|
return out
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Loading and small parsers
|
|
# --------------------------------------------------------------------------- #
|
|
def strip_line_comments(text: str) -> str:
|
|
"""Remove whole-line `//` comments so an annotated template loads as JSON.
|
|
|
|
Only lines whose first non-whitespace characters are `//` are dropped. This
|
|
is deliberately conservative: it never touches `//` inside a string value
|
|
(a source URL such as ``https://...`` is mid-line, never at column zero after
|
|
trimming), and it never removes trailing commas, so it cannot silently
|
|
corrupt an otherwise-valid file.
|
|
"""
|
|
kept = []
|
|
for line in text.splitlines():
|
|
if line.lstrip().startswith("//"):
|
|
continue
|
|
kept.append(line)
|
|
return "\n".join(kept)
|
|
|
|
|
|
def load_intake(path: str) -> Dict[str, Any]:
|
|
"""Read and parse the intake file, raising ValueError with a clear message.
|
|
|
|
Every failure mode (missing file, unreadable file, invalid JSON, wrong
|
|
top-level type) is turned into a ValueError carrying a human sentence, so the
|
|
caller can print one clean line instead of a traceback.
|
|
"""
|
|
try:
|
|
with open(path, "r", encoding="utf-8-sig") as handle:
|
|
raw = handle.read()
|
|
except FileNotFoundError:
|
|
raise ValueError("cannot read %r: no such file" % path)
|
|
except OSError as err:
|
|
raise ValueError("cannot read %r: %s" % (path, err.strerror or err))
|
|
|
|
try:
|
|
doc = json.loads(strip_line_comments(raw))
|
|
except json.JSONDecodeError as err:
|
|
raise ValueError(
|
|
"%s is not valid JSON: %s (line %d, column %d)"
|
|
% (path, err.msg, err.lineno, err.colno)
|
|
)
|
|
|
|
if not isinstance(doc, dict):
|
|
raise ValueError(
|
|
"%s must contain a JSON object at the top level, got %s"
|
|
% (path, type(doc).__name__)
|
|
)
|
|
return doc
|
|
|
|
|
|
def to_number(value: Any) -> Optional[float]:
|
|
"""Coerce a datapoint value to float, or return None if it is not numeric.
|
|
|
|
Accepts ints/floats and numeric strings with thousands separators or a
|
|
leading currency symbol ("1,23,456", "₹ 59,500"). Booleans and free text
|
|
return None so they are simply skipped by numeric checks rather than
|
|
crashing them.
|
|
"""
|
|
if isinstance(value, bool):
|
|
return None
|
|
if isinstance(value, (int, float)):
|
|
return float(value)
|
|
if isinstance(value, str):
|
|
cleaned = re.sub(r"[,\s₹$£€¥]", "", value)
|
|
cleaned = re.sub(r"(?i)(cr|crore|mn|million|bn|billion|lakh|k)$", "", cleaned)
|
|
try:
|
|
return float(cleaned)
|
|
except ValueError:
|
|
return None
|
|
return None
|
|
|
|
|
|
def parse_iso_date(value: Any) -> Optional[date]:
|
|
"""Parse an ISO ``YYYY-MM-DD`` string to a date, or None if unparseable."""
|
|
if not isinstance(value, str):
|
|
return None
|
|
try:
|
|
return date.fromisoformat(value.strip()[:10])
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def normalise_basis(value: Any) -> Optional[str]:
|
|
"""Map a basis string to 'consolidated' / 'standalone' / other, or None."""
|
|
if not isinstance(value, str) or not value.strip():
|
|
return None
|
|
low = value.strip().lower()
|
|
if low.startswith("cons"):
|
|
return "consolidated"
|
|
if low.startswith("stand") or low.startswith("sa") or low == "parent":
|
|
return "standalone"
|
|
return low
|
|
|
|
|
|
def resolve_fy_end_month(doc: Dict[str, Any]) -> int:
|
|
"""Return the file's fiscal-year-end month (1-12); default March (Indian).
|
|
|
|
A fiscal-year label like "FY26" carries no month on its face, so converting
|
|
it to a period-end date requires knowing the year-end. The skill is
|
|
India-leaning (crore/INR), where the year ends in March, so that is the
|
|
default; it can be overridden with a top-level ``fiscal_year_end`` field.
|
|
"""
|
|
raw = doc.get("fiscal_year_end")
|
|
if isinstance(raw, (int, float)) and 1 <= int(raw) <= 12:
|
|
return int(raw)
|
|
if isinstance(raw, str):
|
|
key = raw.strip().lower()[:3]
|
|
if key in _MONTHS:
|
|
return _MONTHS[key]
|
|
return 3
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Unit parsing
|
|
# --------------------------------------------------------------------------- #
|
|
@dataclass
|
|
class ParsedUnit:
|
|
"""Decomposition of a unit string into currency, scale and monetary-ness."""
|
|
|
|
currency: Optional[str]
|
|
scale: Optional[str]
|
|
monetary: bool
|
|
raw: str
|
|
|
|
|
|
def parse_unit(unit: Any) -> Optional[ParsedUnit]:
|
|
"""Split a unit string into a currency code and a scale name.
|
|
|
|
Returns None when there is nothing to parse. ``monetary`` is False for
|
|
percentages, multiples, day-counts and the like -- those describe ratios, not
|
|
amounts, and must be exempt from the currency/units comparison and the 10x
|
|
magnitude test.
|
|
"""
|
|
if not isinstance(unit, str) or not unit.strip():
|
|
return None
|
|
raw = unit.strip()
|
|
tokens = re.split(r"[\s/,]+", raw.lower())
|
|
currency: Optional[str] = None
|
|
scale: Optional[str] = None
|
|
non_monetary = False
|
|
for tok in tokens:
|
|
if not tok:
|
|
continue
|
|
if tok in _CURRENCY_ALIASES and currency is None:
|
|
currency = _CURRENCY_ALIASES[tok]
|
|
elif tok in _SCALE_ALIASES and scale is None:
|
|
scale = _SCALE_ALIASES[tok]
|
|
elif tok in _NON_MONETARY_TOKENS or "%" in tok:
|
|
non_monetary = True
|
|
monetary = (currency is not None or scale is not None) and not non_monetary
|
|
return ParsedUnit(currency=currency, scale=scale, monetary=monetary, raw=raw)
|
|
|
|
|
|
def normalise_currency(value: Any) -> Optional[str]:
|
|
"""Map a bare currency declaration to a canonical code."""
|
|
if not isinstance(value, str) or not value.strip():
|
|
return None
|
|
key = value.strip().lower()
|
|
return _CURRENCY_ALIASES.get(key, value.strip().upper())
|
|
|
|
|
|
def normalise_scale(value: Any) -> Optional[str]:
|
|
"""Map a bare scale declaration to a canonical name."""
|
|
if not isinstance(value, str) or not value.strip():
|
|
return None
|
|
key = value.strip().lower()
|
|
return _SCALE_ALIASES.get(key, key)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Period parsing
|
|
# --------------------------------------------------------------------------- #
|
|
@dataclass
|
|
class ParsedPeriod:
|
|
"""A period label resolved to a kind, a fiscal-year-end month and an end date.
|
|
|
|
kind -- one of: fiscal_year, quarter, calendar_year, ttm,
|
|
bare_year, unknown.
|
|
fy_end_month -- the resolved year-end month (explicit if the label carried
|
|
one, else the file default).
|
|
end_date -- approximate period-end, used only for chronological ordering
|
|
and staleness math; None when it cannot be inferred.
|
|
ambiguous -- True for a bare "2025" that could be FY or CY.
|
|
"""
|
|
|
|
kind: str
|
|
fy_end_month: Optional[int]
|
|
end_date: Optional[date]
|
|
ambiguous: bool = False
|
|
|
|
|
|
def _last_day_of_month(year: int, month: int) -> date:
|
|
"""Return the last calendar day of the given month."""
|
|
if month == 12:
|
|
return date(year, 12, 31)
|
|
return date(year, month + 1, 1) - timedelta(days=1)
|
|
|
|
|
|
def _shift_months(year: int, month: int, delta: int) -> Tuple[int, int]:
|
|
"""Return (year, month) shifted by ``delta`` months (delta may be negative)."""
|
|
total = year * 12 + (month - 1) + delta
|
|
return total // 12, total % 12 + 1
|
|
|
|
|
|
def _fy_year(token: str) -> int:
|
|
"""Turn a fiscal-year token ('27', '2027') into a 4-digit year."""
|
|
digits = int(token)
|
|
if len(token) <= 2:
|
|
return 2000 + digits
|
|
return digits
|
|
|
|
|
|
def _find_month_and_year(text: str) -> Tuple[Optional[int], Optional[int]]:
|
|
"""Extract a (month, year) pair from free text, if both are present."""
|
|
month = None
|
|
for name, num in _MONTHS.items():
|
|
if re.search(r"\b" + name + r"\b", text, re.IGNORECASE):
|
|
month = num
|
|
break
|
|
ym = re.search(r"((?:19|20)\d{2})", text)
|
|
year = int(ym.group(1)) if ym else None
|
|
return month, year
|
|
|
|
|
|
def classify_period(label: Any, default_fy_end_month: int) -> ParsedPeriod:
|
|
"""Classify a period label and compute an approximate period-end date.
|
|
|
|
Handles the common Indian and global forms: "FY26", "Q1 FY27", "CY2025",
|
|
"TTM to Sep-2025", "year ended Dec-2025", and a bare "2025". Anything it
|
|
cannot place is returned as kind 'unknown' with no end date -- degraded, never
|
|
a crash.
|
|
"""
|
|
if not isinstance(label, str) or not label.strip():
|
|
return ParsedPeriod("unknown", None, None)
|
|
up = label.upper()
|
|
|
|
# Trailing-twelve-months: a rolling window, not a discrete fiscal period.
|
|
if "TTM" in up or "LTM" in up:
|
|
month, year = _find_month_and_year(label)
|
|
end = _last_day_of_month(year, month) if (month and year) else None
|
|
return ParsedPeriod("ttm", month, end)
|
|
|
|
# Quarter, fiscal ("Q1 FY27") or calendar ("Q1 2027" / "Q1 CY2027").
|
|
qm = re.search(r"Q\s*([1-4])", up)
|
|
if qm:
|
|
quarter = int(qm.group(1))
|
|
fy = re.search(r"FY\s*([0-9]{2,4})", up)
|
|
if fy:
|
|
year = _fy_year(fy.group(1))
|
|
fem = default_fy_end_month
|
|
end = _quarter_end(year, fem, quarter)
|
|
return ParsedPeriod("quarter", fem, end)
|
|
cy = re.search(r"(?:CY\s*)?((?:19|20)\d{2})", up)
|
|
if cy:
|
|
year = int(cy.group(1))
|
|
end = _quarter_end(year, 12, quarter)
|
|
return ParsedPeriod("quarter", 12, end)
|
|
return ParsedPeriod("quarter", None, None)
|
|
|
|
# Fiscal year: "FY26", "FY2026".
|
|
fy = re.search(r"FY\s*([0-9]{2,4})", up)
|
|
if fy:
|
|
year = _fy_year(fy.group(1))
|
|
# An explicit month token overrides the default year-end.
|
|
month, _ = _find_month_and_year(label)
|
|
fem = month if month else default_fy_end_month
|
|
return ParsedPeriod("fiscal_year", fem, _last_day_of_month(year, fem))
|
|
|
|
# Calendar year: "CY2025".
|
|
cy = re.search(r"CY\s*([0-9]{2,4})", up)
|
|
if cy:
|
|
year = _fy_year(cy.group(1))
|
|
return ParsedPeriod("calendar_year", 12, _last_day_of_month(year, 12))
|
|
|
|
# Explicit month-year: "Mar-2026", "year ended Dec-2025", "December 2025".
|
|
month, year = _find_month_and_year(label)
|
|
if month and year:
|
|
return ParsedPeriod("fiscal_year", month, _last_day_of_month(year, month))
|
|
|
|
# Bare year: "2025" -- genuinely ambiguous between FY and CY.
|
|
bare = re.fullmatch(r"\s*((?:19|20)\d{2})\s*", label)
|
|
if bare:
|
|
year = int(bare.group(1))
|
|
return ParsedPeriod("bare_year", 12, _last_day_of_month(year, 12), ambiguous=True)
|
|
|
|
return ParsedPeriod("unknown", None, None)
|
|
|
|
|
|
def _quarter_end(fy_year: int, fy_end_month: int, quarter: int) -> date:
|
|
"""Return the period-end date of fiscal quarter ``quarter``.
|
|
|
|
Q4 ends on the fiscal-year-end; each earlier quarter ends three months
|
|
before the next. With a March year-end, FY27 Q1 ends 30-Jun-2026, exactly the
|
|
conversion the recency check needs to notice a gathered FY figure is a
|
|
quarter behind the latest reported quarter.
|
|
"""
|
|
months_before_year_end = 3 * (4 - quarter)
|
|
year, month = _shift_months(fy_year, fy_end_month, -months_before_year_end)
|
|
return _last_day_of_month(year, month)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Datapoint helpers
|
|
# --------------------------------------------------------------------------- #
|
|
def is_unavailable(dp: Dict[str, Any]) -> bool:
|
|
"""True if the datapoint is an explicit 'not available' marker, not a figure."""
|
|
value = dp.get("value", None)
|
|
if value is None and "value" in dp:
|
|
return True
|
|
if isinstance(value, str) and _UNAVAILABLE_RE.match(value):
|
|
return True
|
|
return False
|
|
|
|
|
|
def dp_entity(dp: Dict[str, Any], company: Optional[str]) -> str:
|
|
"""Return whose figure this is -- the named entity, else the subject company."""
|
|
ent = dp.get("entity")
|
|
if isinstance(ent, str) and ent.strip():
|
|
return ent.strip()
|
|
return company or "(company)"
|
|
|
|
|
|
def dp_label(dp: Dict[str, Any], company: Optional[str]) -> str:
|
|
"""Build a stable, human descriptor of a datapoint for finding messages."""
|
|
metric = dp.get("metric") or "(unnamed metric)"
|
|
period = dp.get("period") or "no period"
|
|
entity = dp_entity(dp, company)
|
|
if company and entity != company:
|
|
return "%s / %s (%s)" % (entity, metric, period)
|
|
return "%s (%s)" % (metric, period)
|
|
|
|
|
|
def infer_source_tier(source: Any) -> Optional[int]:
|
|
"""Infer a source tier (1-4) from a free-text source string, or None.
|
|
|
|
Highest tier wins: a source that names a primary filing is Tier 1 even if it
|
|
also mentions an aggregator, because that means the figure was traced to the
|
|
filing and merely cross-checked. A source that names only an aggregator is
|
|
Tier 4. Anything the vocabulary does not recognise returns None ('unknown'),
|
|
which the caller treats as unconfirmed rather than assuming the best.
|
|
"""
|
|
if not isinstance(source, str) or not source.strip():
|
|
return None
|
|
if _TIER1_SOURCE_RE.search(source):
|
|
return 1
|
|
if _TIER2_SOURCE_RE.search(source):
|
|
return 2
|
|
if _TIER3_SOURCE_RE.search(source):
|
|
return 3
|
|
if _TIER4_SOURCE_RE.search(source):
|
|
return 4
|
|
return None
|
|
|
|
|
|
def resolve_source_tier(dp: Dict[str, Any]) -> Tuple[Optional[int], bool]:
|
|
"""Return (tier, inferred). An explicit `source_tier` wins over inference.
|
|
|
|
``inferred`` is True when the tier came from reading the source text rather
|
|
than from an explicit field, so the caller can phrase findings honestly.
|
|
"""
|
|
raw = dp.get("source_tier")
|
|
if isinstance(raw, bool):
|
|
raw = None
|
|
if isinstance(raw, (int, float)) and int(raw) in (1, 2, 3, 4):
|
|
return int(raw), False
|
|
if isinstance(raw, str):
|
|
m = re.search(r"[1-4]", raw)
|
|
if m:
|
|
return int(m.group(0)), False
|
|
return infer_source_tier(dp.get("source")), True
|
|
|
|
|
|
def _normalise_metric(metric: str) -> str:
|
|
"""Turn snake_case / kebab-case metric names into space-separated words.
|
|
|
|
Canonical metric names in the intake are snake_case ("net_profit",
|
|
"operating_cash_flow", "market_cap"), so the word-oriented patterns must see
|
|
"net profit", not "net_profit" -- otherwise every headline metric slips the
|
|
check and the market-cap exemption never fires.
|
|
"""
|
|
return re.sub(r"[_\-]+", " ", metric)
|
|
|
|
|
|
def is_headline_metric(metric: Any) -> bool:
|
|
"""True for the load-bearing metrics whose source tier is enforced hardest."""
|
|
if not isinstance(metric, str):
|
|
return False
|
|
norm = _normalise_metric(metric)
|
|
if _MARKET_DATA_METRIC_RE.search(norm):
|
|
return False
|
|
return bool(_HEADLINE_METRIC_RE.search(norm))
|
|
|
|
|
|
def is_market_data_metric(metric: Any) -> bool:
|
|
"""True for price/market-cap style metrics -- exempt from the tier check."""
|
|
if not isinstance(metric, str):
|
|
return False
|
|
return bool(_MARKET_DATA_METRIC_RE.search(_normalise_metric(metric)))
|
|
|
|
|
|
def _iter_real_datapoints(
|
|
doc: Dict[str, Any]
|
|
) -> List[Tuple[int, Dict[str, Any]]]:
|
|
"""Yield (index, datapoint) for well-formed datapoints that carry a value.
|
|
|
|
'Not available' markers and malformed entries are excluded here; malformed
|
|
entries are reported separately by :func:`check_datapoint_shape`.
|
|
"""
|
|
out = []
|
|
for idx, dp in enumerate(doc.get("datapoints", []) or []):
|
|
if not isinstance(dp, dict):
|
|
continue
|
|
if is_unavailable(dp):
|
|
continue
|
|
out.append((idx, dp))
|
|
return out
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Structural / metadata checks
|
|
# --------------------------------------------------------------------------- #
|
|
def check_datapoint_shape(doc: Dict[str, Any]) -> List[Finding]:
|
|
"""Flag datapoints that are not objects or lack a metric name (error).
|
|
|
|
A datapoint with no metric cannot be grouped, cited or reasoned about, so it
|
|
is a hard error rather than a skipped row -- silently dropping it would hide
|
|
data the analyst believes is present.
|
|
"""
|
|
findings: List[Finding] = []
|
|
datapoints = doc.get("datapoints", [])
|
|
if not isinstance(datapoints, list) or not datapoints:
|
|
findings.append(
|
|
Finding(
|
|
"NO_DATAPOINTS",
|
|
SEVERITY_ERROR,
|
|
"No datapoints to verify: the 'datapoints' array is missing or empty.",
|
|
)
|
|
)
|
|
return findings
|
|
for idx, dp in enumerate(datapoints):
|
|
if not isinstance(dp, dict):
|
|
findings.append(
|
|
Finding(
|
|
"MALFORMED_DATAPOINT",
|
|
SEVERITY_ERROR,
|
|
"Datapoint #%d is not an object (got %s)."
|
|
% (idx, type(dp).__name__),
|
|
)
|
|
)
|
|
continue
|
|
metric = dp.get("metric")
|
|
if not isinstance(metric, str) or not metric.strip():
|
|
findings.append(
|
|
Finding(
|
|
"MALFORMED_DATAPOINT",
|
|
SEVERITY_ERROR,
|
|
"Datapoint #%d has no 'metric' name; it cannot be cited or grouped."
|
|
% idx,
|
|
)
|
|
)
|
|
return findings
|
|
|
|
|
|
def check_metadata(doc: Dict[str, Any]) -> List[Finding]:
|
|
"""Flag missing file-level declarations that weaken downstream checks (warn).
|
|
|
|
None of these are fatal on their own, but each disables or narrows a later
|
|
check -- a missing as_of makes staleness unmeasurable, a missing currency
|
|
makes unit mixing invisible -- so the analyst is told which guarantees the
|
|
gate could not provide.
|
|
"""
|
|
findings: List[Finding] = []
|
|
if not (isinstance(doc.get("company"), str) and doc["company"].strip()):
|
|
findings.append(
|
|
Finding("META_MISSING_COMPANY", SEVERITY_WARN,
|
|
"File-level 'company' is missing; datapoints cannot be attributed.")
|
|
)
|
|
if parse_iso_date(doc.get("as_of")) is None:
|
|
findings.append(
|
|
Finding("META_MISSING_AS_OF", SEVERITY_WARN,
|
|
"File-level 'as_of' is missing or not an ISO date; "
|
|
"staleness/recency checks are skipped.")
|
|
)
|
|
if normalise_basis(doc.get("reporting_basis")) is None:
|
|
findings.append(
|
|
Finding("META_MISSING_BASIS", SEVERITY_WARN,
|
|
"File-level 'reporting_basis' is missing; "
|
|
"cannot tell which basis the analysis is meant to run on.")
|
|
)
|
|
if normalise_currency(doc.get("currency")) is None:
|
|
findings.append(
|
|
Finding("META_MISSING_CURRENCY", SEVERITY_WARN,
|
|
"File-level 'currency' is missing; currency mixing cannot be detected.")
|
|
)
|
|
if normalise_scale(doc.get("units")) is None:
|
|
findings.append(
|
|
Finding("META_MISSING_UNITS", SEVERITY_WARN,
|
|
"File-level 'units' is missing; scale mixing (the 10x error) "
|
|
"cannot be detected.")
|
|
)
|
|
return findings
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Check 1 -- provenance
|
|
# --------------------------------------------------------------------------- #
|
|
def check_provenance(doc: Dict[str, Any]) -> List[Finding]:
|
|
"""Every real figure must be traceable: source, period, unit, basis.
|
|
|
|
Source and period are the two fields without which a number cannot be
|
|
re-found or placed in time, so their absence is an error. Unit and basis can
|
|
often be inferred from the file-level declaration, so their absence is a
|
|
warning -- but an unstated basis is exactly how consolidated and standalone
|
|
get mixed, so it is never silent.
|
|
"""
|
|
company = doc.get("company")
|
|
findings: List[Finding] = []
|
|
for _, dp in _iter_real_datapoints(doc):
|
|
label = dp_label(dp, company)
|
|
metric = dp.get("metric")
|
|
entity = dp_entity(dp, company)
|
|
period = dp.get("period")
|
|
if not (isinstance(dp.get("source"), str) and dp["source"].strip()):
|
|
findings.append(
|
|
Finding("PROVENANCE_MISSING_SOURCE", SEVERITY_ERROR,
|
|
"%s has no source; the figure cannot be traced or re-verified."
|
|
% label, metric, period, entity)
|
|
)
|
|
if not (isinstance(period, str) and period.strip()):
|
|
findings.append(
|
|
Finding("PROVENANCE_MISSING_PERIOD", SEVERITY_ERROR,
|
|
"%s has no period; a figure without a period is not yet a fact."
|
|
% label, metric, period, entity)
|
|
)
|
|
if not (isinstance(dp.get("unit"), str) and dp["unit"].strip()):
|
|
findings.append(
|
|
Finding("PROVENANCE_MISSING_UNIT", SEVERITY_WARN,
|
|
"%s has no unit; scale/currency must be inferred from the "
|
|
"file declaration." % label, metric, period, entity)
|
|
)
|
|
if normalise_basis(dp.get("basis")) is None:
|
|
findings.append(
|
|
Finding("PROVENANCE_MISSING_BASIS", SEVERITY_WARN,
|
|
"%s has no basis; consolidated vs standalone is left implicit."
|
|
% label, metric, period, entity)
|
|
)
|
|
return findings
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Check 2 -- cross-source disagreement
|
|
# --------------------------------------------------------------------------- #
|
|
def check_cross_source(doc: Dict[str, Any]) -> List[Finding]:
|
|
"""Compare every ``alt`` read against the primary ``value``.
|
|
|
|
This is the check that would have caught the wrong-NPA case: two sources for
|
|
the same figure that disagree by more than a rounding difference. The
|
|
divergence is reported with BOTH figures and BOTH sources so the analyst
|
|
reconciles rather than averages -- averaging two numbers you cannot reconcile
|
|
is precisely the error that puts a wrong figure into the report.
|
|
"""
|
|
company = doc.get("company")
|
|
findings: List[Finding] = []
|
|
for _, dp in _iter_real_datapoints(doc):
|
|
alts = dp.get("alt")
|
|
if not isinstance(alts, list) or not alts:
|
|
continue
|
|
primary = to_number(dp.get("value"))
|
|
if primary is None:
|
|
continue
|
|
label = dp_label(dp, company)
|
|
metric = dp.get("metric")
|
|
entity = dp_entity(dp, company)
|
|
period = dp.get("period")
|
|
primary_src = dp.get("source") or "primary source not given"
|
|
for alt in alts:
|
|
if not isinstance(alt, dict):
|
|
continue
|
|
alt_val = to_number(alt.get("value"))
|
|
if alt_val is None:
|
|
continue
|
|
alt_src = alt.get("source") or "alt source not given"
|
|
if primary == 0.0:
|
|
pct = 0.0 if alt_val == 0.0 else float("inf")
|
|
else:
|
|
pct = abs(alt_val - primary) / abs(primary) * 100.0
|
|
|
|
if pct == float("inf"):
|
|
findings.append(
|
|
Finding("CROSS_SOURCE_DIVERGENCE", SEVERITY_ERROR,
|
|
"%s: primary is 0 (%s) but a source reports %s (%s); "
|
|
"reconcile before use." % (
|
|
label, primary_src, _fmt(alt_val), alt_src),
|
|
metric, period, entity))
|
|
continue
|
|
if pct <= DIVERGENCE_WARN_PCT:
|
|
continue
|
|
severity = SEVERITY_ERROR if pct > DIVERGENCE_ERROR_PCT else SEVERITY_WARN
|
|
findings.append(
|
|
Finding("CROSS_SOURCE_DIVERGENCE", severity,
|
|
"%s: sources disagree by %.1f%% -- %s (%s) vs %s (%s). %s"
|
|
% (label, pct, _fmt(primary), primary_src,
|
|
_fmt(alt_val), alt_src,
|
|
"Different basis/period/unit is likely; reconcile before use."
|
|
if severity == SEVERITY_ERROR else
|
|
"Probably a definitional difference; confirm which to use."),
|
|
metric, period, entity))
|
|
return findings
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Check 3 -- basis mixing
|
|
# --------------------------------------------------------------------------- #
|
|
def check_basis_mixing(doc: Dict[str, Any]) -> List[Finding]:
|
|
"""Detect consolidated and standalone figures mixed in one dataset.
|
|
|
|
A ratio whose numerator is consolidated and denominator standalone looks
|
|
plausible and is meaningless, so mixing the two bases within an analysis is an
|
|
error -- unless each deviating figure carries an explicit note explaining why
|
|
(e.g. standalone debt because subsidiary debt is non-recourse), in which case
|
|
it is a disclosed choice and downgraded to a warning. A dataset whose figures
|
|
all contradict the declared basis is flagged too, because the declaration is
|
|
then almost certainly wrong.
|
|
"""
|
|
company = doc.get("company")
|
|
declared = normalise_basis(doc.get("reporting_basis"))
|
|
reals = _iter_real_datapoints(doc)
|
|
bases = {} # basis -> list of (dp)
|
|
for _, dp in reals:
|
|
b = normalise_basis(dp.get("basis"))
|
|
if b in ("consolidated", "standalone"):
|
|
bases.setdefault(b, []).append(dp)
|
|
|
|
present = set(bases)
|
|
findings: List[Finding] = []
|
|
if "consolidated" in present and "standalone" in present:
|
|
# Offenders are the figures that deviate from the declared basis; if no
|
|
# basis is declared, the smaller group is treated as the intrusion.
|
|
if declared in present:
|
|
offender_basis = "standalone" if declared == "consolidated" else "consolidated"
|
|
else:
|
|
offender_basis = min(present, key=lambda b: len(bases[b]))
|
|
offenders = bases[offender_basis]
|
|
unnoted = [dp for dp in offenders
|
|
if not (isinstance(dp.get("note"), str) and dp["note"].strip())]
|
|
keeper_basis = "consolidated" if offender_basis == "standalone" else "standalone"
|
|
off_labels = ", ".join(dp_label(dp, company) for dp in offenders)
|
|
keep_labels = ", ".join(dp_label(dp, company) for dp in bases[keeper_basis])
|
|
if unnoted:
|
|
findings.append(
|
|
Finding("BASIS_MIXING", SEVERITY_ERROR,
|
|
"Dataset silently mixes bases: %s on %s vs %s on %s. "
|
|
"Any ratio spanning both is invalid; put the whole analysis "
|
|
"on one basis or note each exception."
|
|
% (off_labels, offender_basis, keep_labels, keeper_basis)))
|
|
else:
|
|
findings.append(
|
|
Finding("BASIS_MIXING_NOTED", SEVERITY_WARN,
|
|
"Dataset mixes bases but each exception is noted (%s on %s). "
|
|
"Confirm no single ratio draws inputs from both bases."
|
|
% (off_labels, offender_basis)))
|
|
elif declared and present and declared not in present:
|
|
only = next(iter(present))
|
|
findings.append(
|
|
Finding("BASIS_DECLARATION_MISMATCH", SEVERITY_WARN,
|
|
"File declares '%s' basis but every figure is '%s'; "
|
|
"the declared basis appears wrong." % (declared, only)))
|
|
return findings
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Check 4 -- unit / currency mixing
|
|
# --------------------------------------------------------------------------- #
|
|
def check_unit_currency(doc: Dict[str, Any]) -> List[Finding]:
|
|
"""Flag unit/currency drift from the file declaration and 10x magnitude slips.
|
|
|
|
Two distinct failures share this check. First, a figure labelled in a
|
|
different currency or scale than the file declares, without a note -- the
|
|
quiet path to the crore-vs-million 10x error. Second, two supposedly
|
|
comparable figures (a value against its own alt, or two reads of the same
|
|
metric+period) whose ratio sits almost exactly on 10x/100x -- the loud
|
|
signature of a scale slip, reported as an error because a clean power of ten
|
|
is never a real difference between the same quantity.
|
|
"""
|
|
company = doc.get("company")
|
|
file_currency = normalise_currency(doc.get("currency"))
|
|
file_scale = normalise_scale(doc.get("units"))
|
|
findings: List[Finding] = []
|
|
|
|
# Part A -- label drift from the file declaration.
|
|
for _, dp in _iter_real_datapoints(doc):
|
|
parsed = parse_unit(dp.get("unit"))
|
|
if parsed is None or not parsed.monetary:
|
|
continue
|
|
noted = isinstance(dp.get("note"), str) and dp["note"].strip()
|
|
if noted:
|
|
continue
|
|
label = dp_label(dp, company)
|
|
metric = dp.get("metric")
|
|
entity = dp_entity(dp, company)
|
|
period = dp.get("period")
|
|
if file_currency and parsed.currency and parsed.currency != file_currency:
|
|
findings.append(
|
|
Finding("UNIT_CURRENCY_MISMATCH", SEVERITY_WARN,
|
|
"%s is in %s but the file declares %s; add a note or "
|
|
"convert (state the FX rate and date)."
|
|
% (label, parsed.currency, file_currency),
|
|
metric, period, entity))
|
|
if file_scale and parsed.scale and parsed.scale != file_scale:
|
|
findings.append(
|
|
Finding("UNIT_SCALE_MISMATCH", SEVERITY_WARN,
|
|
"%s is in %s but the file declares %s; a silent crore/million "
|
|
"mismatch is a 10x error." % (label, parsed.scale, file_scale),
|
|
metric, period, entity))
|
|
|
|
# Part B -- magnitude slips within comparable figures.
|
|
findings.extend(_check_magnitude(doc))
|
|
return findings
|
|
|
|
|
|
def _is_power_of_ten_gap(a: float, b: float) -> Optional[float]:
|
|
"""Return the ~10^k factor separating a and b, if one does within tolerance.
|
|
|
|
Only clean 10x/100x/1000x gaps qualify: two reads of the same quantity never
|
|
differ by an exact order of magnitude for a legitimate reason, so such a gap
|
|
is a scale error rather than a real difference.
|
|
"""
|
|
if a <= 0 or b <= 0:
|
|
return None
|
|
hi, lo = (a, b) if a >= b else (b, a)
|
|
ratio = hi / lo
|
|
for factor in MAGNITUDE_FACTORS:
|
|
if abs(ratio / factor - 1.0) <= MAGNITUDE_TOLERANCE:
|
|
return factor
|
|
return None
|
|
|
|
|
|
def _check_magnitude(doc: Dict[str, Any]) -> List[Finding]:
|
|
"""Find order-of-magnitude gaps between comparable monetary figures."""
|
|
company = doc.get("company")
|
|
findings: List[Finding] = []
|
|
|
|
# (i) value vs its own alt reads.
|
|
for _, dp in _iter_real_datapoints(doc):
|
|
parsed = parse_unit(dp.get("unit"))
|
|
if parsed is not None and not parsed.monetary:
|
|
continue
|
|
primary = to_number(dp.get("value"))
|
|
if primary is None:
|
|
continue
|
|
alts = dp.get("alt")
|
|
if not isinstance(alts, list):
|
|
continue
|
|
for alt in alts:
|
|
if not isinstance(alt, dict):
|
|
continue
|
|
alt_val = to_number(alt.get("value"))
|
|
if alt_val is None:
|
|
continue
|
|
factor = _is_power_of_ten_gap(primary, alt_val)
|
|
if factor:
|
|
findings.append(
|
|
Finding("UNIT_MAGNITUDE_MISMATCH", SEVERITY_ERROR,
|
|
"%s: primary %s and source '%s' %s differ by ~%dx -- "
|
|
"a unit/scale error (e.g. crore vs million), not a real gap."
|
|
% (dp_label(dp, company), _fmt(primary),
|
|
alt.get("source") or "?", _fmt(alt_val), int(factor)),
|
|
dp.get("metric"), dp.get("period"),
|
|
dp_entity(dp, company)))
|
|
|
|
# (ii) two datapoints for the same entity+metric+period.
|
|
groups: Dict[Tuple[str, str, str], List[Dict[str, Any]]] = {}
|
|
for _, dp in _iter_real_datapoints(doc):
|
|
parsed = parse_unit(dp.get("unit"))
|
|
if parsed is not None and not parsed.monetary:
|
|
continue
|
|
if to_number(dp.get("value")) is None:
|
|
continue
|
|
key = (dp_entity(dp, company), str(dp.get("metric")),
|
|
str(dp.get("period")))
|
|
groups.setdefault(key, []).append(dp)
|
|
for dps in groups.values():
|
|
for i in range(len(dps)):
|
|
for j in range(i + 1, len(dps)):
|
|
a = to_number(dps[i].get("value"))
|
|
b = to_number(dps[j].get("value"))
|
|
factor = _is_power_of_ten_gap(a, b)
|
|
if factor:
|
|
findings.append(
|
|
Finding("UNIT_MAGNITUDE_MISMATCH", SEVERITY_ERROR,
|
|
"%s and %s are the same metric and period yet differ "
|
|
"by ~%dx -- a likely unit/scale error."
|
|
% (dp_label(dps[i], company),
|
|
dp_label(dps[j], company), int(factor)),
|
|
dps[i].get("metric"), dps[i].get("period"),
|
|
dp_entity(dps[i], company)))
|
|
return findings
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Check 5 -- period alignment
|
|
# --------------------------------------------------------------------------- #
|
|
def check_period_alignment(doc: Dict[str, Any]) -> List[Finding]:
|
|
"""Warn when figures for one metric mix incompatible period conventions.
|
|
|
|
Figures compared as like-for-like must share a period convention. A rolling
|
|
TTM window is not the same shape as a discrete fiscal year; a calendar year
|
|
sits ~3 months off an Indian fiscal year; a March year-end and a December
|
|
year-end are not the same year. Having both an annual and a quarterly figure
|
|
for a metric is normal and is NOT flagged -- only genuinely incomparable mixes
|
|
are.
|
|
"""
|
|
company = doc.get("company")
|
|
default_fem = resolve_fy_end_month(doc)
|
|
findings: List[Finding] = []
|
|
|
|
groups: Dict[Tuple[str, str], List[Tuple[Dict[str, Any], ParsedPeriod]]] = {}
|
|
for _, dp in _iter_real_datapoints(doc):
|
|
parsed = classify_period(dp.get("period"), default_fem)
|
|
if parsed.kind == "unknown":
|
|
continue
|
|
key = (dp_entity(dp, company), str(dp.get("metric")))
|
|
groups.setdefault(key, []).append((dp, parsed))
|
|
|
|
for (entity, metric), members in groups.items():
|
|
if len(members) < 2:
|
|
continue
|
|
kinds = {p.kind for _, p in members}
|
|
labels = ", ".join(dp_label(dp, company) for dp, _ in members)
|
|
|
|
if "ttm" in kinds and kinds - {"ttm"}:
|
|
findings.append(
|
|
Finding("PERIOD_ALIGNMENT", SEVERITY_WARN,
|
|
"'%s' mixes a rolling TTM window with discrete-period "
|
|
"figures (%s); a TTM is not comparable to a fiscal/calendar "
|
|
"year." % (metric, labels), metric, None,
|
|
None if entity == company else entity))
|
|
if "calendar_year" in kinds and "fiscal_year" in kinds:
|
|
findings.append(
|
|
Finding("PERIOD_ALIGNMENT", SEVERITY_WARN,
|
|
"'%s' mixes calendar-year and fiscal-year periods (%s); "
|
|
"there is a ~3-month offset between them."
|
|
% (metric, labels), metric, None,
|
|
None if entity == company else entity))
|
|
if "bare_year" in kinds and "fiscal_year" in kinds:
|
|
findings.append(
|
|
Finding("PERIOD_ALIGNMENT", SEVERITY_WARN,
|
|
"'%s' mixes a fiscal-year label with a bare calendar-year "
|
|
"label (%s); the bare year is ambiguous (FY or CY?)."
|
|
% (metric, labels), metric, None,
|
|
None if entity == company else entity))
|
|
fiscal_fems = {p.fy_end_month for _, p in members
|
|
if p.kind == "fiscal_year" and p.fy_end_month}
|
|
if len(fiscal_fems) > 1:
|
|
months = ", ".join(_month_name(m) for m in sorted(fiscal_fems))
|
|
findings.append(
|
|
Finding("PERIOD_ALIGNMENT", SEVERITY_WARN,
|
|
"'%s' mixes fiscal years with different year-ends (%s) across "
|
|
"%s; align the fiscal calendars before comparing."
|
|
% (metric, months, labels), metric, None,
|
|
None if entity == company else entity))
|
|
return findings
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Check 6 -- recency / staleness
|
|
# --------------------------------------------------------------------------- #
|
|
def check_recency(doc: Dict[str, Any]) -> List[Finding]:
|
|
"""Warn when the gathered data is a period behind, or plainly stale.
|
|
|
|
Two failure modes, both drawn from real reports. First: the company has
|
|
reported a newer period than anything gathered -- ``latest_reported_period``
|
|
is ahead of the newest datapoint -- meaning a published quarter was omitted.
|
|
Second: even the newest period known here ends more than ~4 months before
|
|
``as_of``, so a further quarterly result has almost certainly been released
|
|
since and was never checked. All dates come from the file (as_of), never the
|
|
wall clock, so the gate is reproducible.
|
|
"""
|
|
findings: List[Finding] = []
|
|
as_of = parse_iso_date(doc.get("as_of"))
|
|
default_fem = resolve_fy_end_month(doc)
|
|
|
|
# Newest period-end among gathered datapoints.
|
|
dp_ends: List[Tuple[date, Dict[str, Any]]] = []
|
|
for _, dp in _iter_real_datapoints(doc):
|
|
parsed = classify_period(dp.get("period"), default_fem)
|
|
if parsed.end_date is not None:
|
|
dp_ends.append((parsed.end_date, dp))
|
|
newest_dp = max(dp_ends, key=lambda t: t[0]) if dp_ends else None
|
|
|
|
# The latest period the company has actually reported.
|
|
lrp = doc.get("latest_reported_period")
|
|
lrp_end: Optional[date] = None
|
|
lrp_period = None
|
|
lrp_pub = None
|
|
if isinstance(lrp, dict):
|
|
lrp_period = lrp.get("period")
|
|
lrp_end = classify_period(lrp_period, default_fem).end_date
|
|
lrp_pub = parse_iso_date(lrp.get("published"))
|
|
|
|
company = doc.get("company")
|
|
|
|
# Mode 1: a reported period newer than anything gathered.
|
|
if lrp_end is not None and newest_dp is not None:
|
|
gap = (lrp_end - newest_dp[0]).days
|
|
if gap > RECENCY_NEWER_PERIOD_DAYS:
|
|
pub = (" published %s" % lrp_pub.isoformat()) if lrp_pub else ""
|
|
findings.append(
|
|
Finding("RECENCY_UNINCORPORATED", SEVERITY_WARN,
|
|
"A more recent reported period exists but is not incorporated: "
|
|
"latest reported is %s (ends ~%s%s), yet the newest gathered "
|
|
"figure is %s (ends ~%s). Pull the newer period before analysing."
|
|
% (lrp_period, lrp_end.isoformat(), pub,
|
|
dp_label(newest_dp[1], company),
|
|
newest_dp[0].isoformat())))
|
|
|
|
# Mode 2: newest known period-end is far behind as_of.
|
|
reference_ends = [e for e in
|
|
[newest_dp[0] if newest_dp else None, lrp_end]
|
|
if e is not None]
|
|
if as_of is not None and reference_ends:
|
|
reference = max(reference_ends)
|
|
stale_days = (as_of - reference).days
|
|
if stale_days > RECENCY_STALE_DAYS:
|
|
findings.append(
|
|
Finding("RECENCY_STALE", SEVERITY_WARN,
|
|
"The newest period-end known here (%s) is %d days (~%.1f months) "
|
|
"before as_of %s; a further quarter has almost certainly been "
|
|
"reported since -- search for results after %s."
|
|
% (reference.isoformat(), stale_days, stale_days / 30.44,
|
|
as_of.isoformat(), reference.isoformat())))
|
|
|
|
# Consistency: a publish date after the gather date is contradictory.
|
|
if as_of is not None and lrp_pub is not None and lrp_pub > as_of:
|
|
findings.append(
|
|
Finding("RECENCY_DATE_INCONSISTENT", SEVERITY_WARN,
|
|
"latest_reported_period.published (%s) is after as_of (%s); "
|
|
"one of the two dates is wrong."
|
|
% (lrp_pub.isoformat(), as_of.isoformat())))
|
|
return findings
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Check 7 -- coverage (orphan 'not available')
|
|
# --------------------------------------------------------------------------- #
|
|
def check_coverage(doc: Dict[str, Any]) -> List[Finding]:
|
|
"""Report, as info, how many metrics were explicitly marked unavailable.
|
|
|
|
Declaring a metric unsourced is the opposite of a defect: it is the honest
|
|
alternative to fabricating a number to fill a cell. This is surfaced as
|
|
coverage context so the reader can weigh how complete the intake is, never as
|
|
something to fix.
|
|
"""
|
|
marked: List[str] = []
|
|
for dp in doc.get("datapoints", []) or []:
|
|
if isinstance(dp, dict) and is_unavailable(dp):
|
|
marked.append(str(dp.get("metric") or "(unnamed)"))
|
|
extra = doc.get("not_available")
|
|
if isinstance(extra, list):
|
|
for item in extra:
|
|
if isinstance(item, str) and item.strip():
|
|
marked.append(item.strip())
|
|
elif isinstance(item, dict) and item.get("metric"):
|
|
marked.append(str(item["metric"]))
|
|
|
|
if not marked:
|
|
return []
|
|
unique = sorted(set(marked))
|
|
return [
|
|
Finding("COVERAGE_UNAVAILABLE", SEVERITY_INFO,
|
|
"Coverage: %d metric(s) explicitly marked not-available "
|
|
"(good discipline -- gaps disclosed, not fabricated): %s."
|
|
% (len(unique), ", ".join(unique)))
|
|
]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Check 8 -- source tier (documents primary, aggregators navigation-only)
|
|
# --------------------------------------------------------------------------- #
|
|
def check_source_tier(doc: Dict[str, Any]) -> List[Finding]:
|
|
"""Enforce the document-first rule: figures must trace to the filings.
|
|
|
|
The skill's source of record is the company's own filings (Tier 1),
|
|
company-published secondary (Tier 2) and regulator/third-party primary
|
|
(Tier 3). Aggregators -- screener.in, tickertape, yahoo, moneycontrol -- are
|
|
Tier 4: navigation aids and optional cross-checks, never a citable source.
|
|
This is the exact hole a report falls through when it looks well-sourced (a
|
|
period label sits next to every figure) yet every number actually came from
|
|
an aggregator screen. So:
|
|
|
|
* a HEADLINE metric (revenue, PAT, EBITDA, debt, equity, operating cash flow)
|
|
whose only source is Tier 4 is an ERROR -- trace it to the filing;
|
|
* any other aggregator-only figure is a WARN -- fine as a cross-check, but
|
|
replace it with the document before relying on it;
|
|
* a headline figure whose source cannot be confirmed as a filing (unknown
|
|
tier) is a WARN -- tag it or cite the document rather than taking it on
|
|
trust;
|
|
* current price / market cap is the one sanctioned exchange-sourced exception
|
|
and is exempt;
|
|
* a coverage line reports the share of figures that are primary-sourced.
|
|
"""
|
|
company = doc.get("company")
|
|
findings: List[Finding] = []
|
|
n_counted = 0 # figures that are not the market-data exception
|
|
n_primary = 0 # Tier 1-3
|
|
n_aggregator = 0 # Tier 4
|
|
n_unknown = 0 # tier could not be established
|
|
|
|
for _, dp in _iter_real_datapoints(doc):
|
|
metric = dp.get("metric")
|
|
if is_market_data_metric(metric):
|
|
continue # price/market cap: exchange-sourced by nature, exempt.
|
|
label = dp_label(dp, company)
|
|
period = dp.get("period")
|
|
entity = dp_entity(dp, company)
|
|
source_text = dp.get("source") if isinstance(dp.get("source"), str) else ""
|
|
tier, inferred = resolve_source_tier(dp)
|
|
headline = is_headline_metric(metric)
|
|
n_counted += 1
|
|
|
|
# Guard: an explicit Tier 1-3 label that contradicts an aggregator-looking
|
|
# source. Silent mislabelling would defeat the whole check.
|
|
explicit = dp.get("source_tier")
|
|
explicit_primary = (
|
|
isinstance(explicit, (int, float)) and not isinstance(explicit, bool)
|
|
and int(explicit) in (1, 2, 3)
|
|
)
|
|
if (explicit_primary and _TIER4_SOURCE_RE.search(source_text)
|
|
and not (_TIER1_SOURCE_RE.search(source_text)
|
|
or _TIER2_SOURCE_RE.search(source_text)
|
|
or _TIER3_SOURCE_RE.search(source_text))):
|
|
findings.append(
|
|
Finding("SOURCE_TIER_MISLABELLED", SEVERITY_WARN,
|
|
"%s is labelled Tier %d but its source reads like an "
|
|
"aggregator (\"%s\"); confirm it traces to a primary "
|
|
"filing." % (label, int(explicit), source_text),
|
|
metric, period, entity))
|
|
|
|
if tier in (1, 2, 3):
|
|
n_primary += 1
|
|
elif tier == 4:
|
|
n_aggregator += 1
|
|
if headline:
|
|
findings.append(
|
|
Finding("SOURCE_TIER_HEADLINE_AGGREGATOR", SEVERITY_ERROR,
|
|
"%s is a headline metric sourced only from an "
|
|
"aggregator/third-party (\"%s\"). Trace it to the "
|
|
"primary filing -- an aggregator is a navigation aid, "
|
|
"not a source of record." % (label, source_text),
|
|
metric, period, entity))
|
|
else:
|
|
findings.append(
|
|
Finding("SOURCE_TIER_AGGREGATOR", SEVERITY_WARN,
|
|
"%s is sourced only from an aggregator (\"%s\"); fine "
|
|
"as a labelled cross-check, but replace it with the "
|
|
"filing before relying on it." % (label, source_text),
|
|
metric, period, entity))
|
|
else: # unknown tier
|
|
n_unknown += 1
|
|
if headline:
|
|
findings.append(
|
|
Finding("SOURCE_TIER_UNCONFIRMED", SEVERITY_WARN,
|
|
"%s is a headline metric whose source (\"%s\") could "
|
|
"not be confirmed as a primary filing; set source_tier "
|
|
"or cite the document so it is not taken on trust."
|
|
% (label, source_text or "no source"),
|
|
metric, period, entity))
|
|
|
|
if n_counted:
|
|
pct = 100.0 * n_primary / n_counted
|
|
findings.append(
|
|
Finding("SOURCE_TIER_COVERAGE", SEVERITY_INFO,
|
|
"Provenance: %d of %d figures (%.0f%%) trace to a primary "
|
|
"source (Tier 1-3); %d aggregator-only, %d unconfirmed. "
|
|
"Headline metrics should be 100%% primary."
|
|
% (n_primary, n_counted, pct, n_aggregator, n_unknown)))
|
|
return findings
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Orchestration
|
|
# --------------------------------------------------------------------------- #
|
|
_CHECKS = (
|
|
check_datapoint_shape,
|
|
check_metadata,
|
|
check_provenance,
|
|
check_cross_source,
|
|
check_basis_mixing,
|
|
check_unit_currency,
|
|
check_period_alignment,
|
|
check_recency,
|
|
check_coverage,
|
|
check_source_tier,
|
|
)
|
|
|
|
|
|
def verify(doc: Dict[str, Any]) -> List[Finding]:
|
|
"""Run every check and return all findings, sorted for the report.
|
|
|
|
Findings are ordered by severity (errors, then warnings, then info) so the
|
|
report and the exit-code decision both read off one list.
|
|
"""
|
|
findings: List[Finding] = []
|
|
for check in _CHECKS:
|
|
findings.extend(check(doc))
|
|
findings.sort(key=lambda f: (_SEVERITY_RANK.get(f.severity, 9), f.rule_id))
|
|
return findings
|
|
|
|
|
|
def counts(findings: Sequence[Finding]) -> Dict[str, int]:
|
|
"""Tally findings by severity."""
|
|
tally = {SEVERITY_ERROR: 0, SEVERITY_WARN: 0, SEVERITY_INFO: 0}
|
|
for f in findings:
|
|
tally[f.severity] = tally.get(f.severity, 0) + 1
|
|
return tally
|
|
|
|
|
|
def verdict(findings: Sequence[Finding]) -> str:
|
|
"""Return the headline verdict from the finding severities."""
|
|
tally = counts(findings)
|
|
if tally[SEVERITY_ERROR]:
|
|
return "FAIL"
|
|
if tally[SEVERITY_WARN]:
|
|
return "PASS-WITH-WARNINGS"
|
|
return "PASS"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Formatting helpers
|
|
# --------------------------------------------------------------------------- #
|
|
def _fmt(number: float) -> str:
|
|
"""Format a figure without noise: no trailing .0 on whole numbers."""
|
|
if number == int(number):
|
|
return "{:,}".format(int(number))
|
|
return "{:,.4g}".format(number) if abs(number) < 1 else "{:,.2f}".format(number)
|
|
|
|
|
|
def _month_name(month: Optional[int]) -> str:
|
|
"""Return an abbreviated month name for a 1-12 month number."""
|
|
names = ["", "Jan", "Feb", "Mar", "Apr", "May", "Jun",
|
|
"Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]
|
|
if isinstance(month, int) and 1 <= month <= 12:
|
|
return names[month]
|
|
return "?"
|
|
|
|
|
|
def render_text(doc: Dict[str, Any], findings: Sequence[Finding]) -> str:
|
|
"""Render the human report: header, verdict, one-line summary, grouped findings."""
|
|
tally = counts(findings)
|
|
v = verdict(findings)
|
|
n_dp = len([d for d in doc.get("datapoints", []) or []
|
|
if isinstance(d, dict) and not is_unavailable(d)])
|
|
n_na = len([d for d in doc.get("datapoints", []) or []
|
|
if isinstance(d, dict) and is_unavailable(d)])
|
|
|
|
lines: List[str] = []
|
|
lines.append("=" * 72)
|
|
lines.append("DATA-INTAKE VERIFICATION")
|
|
lines.append("=" * 72)
|
|
company = doc.get("company") or "(company not stated)"
|
|
ticker = doc.get("ticker")
|
|
lines.append("Company : %s%s" % (company, " [%s]" % ticker if ticker else ""))
|
|
lines.append("As of : %s" % (doc.get("as_of") or "(not stated)"))
|
|
lines.append("Basis : %s" % (doc.get("reporting_basis") or "(not stated)"))
|
|
lines.append("Units : %s %s"
|
|
% (doc.get("currency") or "?", doc.get("units") or "?"))
|
|
lrp = doc.get("latest_reported_period")
|
|
if isinstance(lrp, dict):
|
|
lines.append("Latest reported: %s (published %s)"
|
|
% (lrp.get("period") or "?", lrp.get("published") or "?"))
|
|
lines.append("Datapoints : %d (+%d marked not-available)" % (n_dp, n_na))
|
|
lines.append("")
|
|
lines.append("VERDICT: %s" % v)
|
|
lines.append("Summary: %d error(s), %d warning(s), %d info."
|
|
% (tally[SEVERITY_ERROR], tally[SEVERITY_WARN], tally[SEVERITY_INFO]))
|
|
|
|
for severity, heading in (
|
|
(SEVERITY_ERROR, "ERRORS (block analysis until resolved)"),
|
|
(SEVERITY_WARN, "WARNINGS (review before relying on the figures)"),
|
|
(SEVERITY_INFO, "INFO (coverage context)"),
|
|
):
|
|
group = [f for f in findings if f.severity == severity]
|
|
if not group:
|
|
continue
|
|
lines.append("")
|
|
lines.append("-" * 72)
|
|
lines.append("%s -- %d" % (heading, len(group)))
|
|
lines.append("-" * 72)
|
|
for f in group:
|
|
lines.append("[%s] %s" % (f.rule_id, f.message))
|
|
|
|
if not findings:
|
|
lines.append("")
|
|
lines.append("No findings: all datapoints are traceable, consistent and current.")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def render_json(doc: Dict[str, Any], findings: Sequence[Finding]) -> str:
|
|
"""Render the machine-readable report used by --json."""
|
|
tally = counts(findings)
|
|
payload = {
|
|
"company": doc.get("company"),
|
|
"ticker": doc.get("ticker"),
|
|
"as_of": doc.get("as_of"),
|
|
"verdict": verdict(findings),
|
|
"counts": {
|
|
"error": tally[SEVERITY_ERROR],
|
|
"warn": tally[SEVERITY_WARN],
|
|
"info": tally[SEVERITY_INFO],
|
|
},
|
|
"findings": [f.to_dict() for f in findings],
|
|
}
|
|
return json.dumps(payload, indent=2, ensure_ascii=False)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Built-in template and example
|
|
# --------------------------------------------------------------------------- #
|
|
TEMPLATE = """\
|
|
// Data-intake verification template for the stock-analysis skill.
|
|
// Fill this in with the data you have GATHERED, then run:
|
|
// python verify_data.py this_file.json
|
|
// Whole-line // comments (like these) are stripped by the loader; keep any
|
|
// comment on its own line. Delete the sample datapoints and add your own.
|
|
{
|
|
"company": "",
|
|
"ticker": "",
|
|
"as_of": "YYYY-MM-DD",
|
|
"reporting_basis": "consolidated",
|
|
"currency": "INR",
|
|
"units": "crore",
|
|
"fiscal_year_end": "March",
|
|
"latest_reported_period": {
|
|
"period": "Q1 FY27",
|
|
"published": "YYYY-MM-DD"
|
|
},
|
|
"datapoints": [
|
|
{
|
|
"metric": "revenue",
|
|
"value": 0,
|
|
"period": "FY26",
|
|
"basis": "consolidated",
|
|
"unit": "INR crore",
|
|
"source": "FY26 annual report, p.xxx",
|
|
// source_tier: 1 filing, 2 company secondary, 3 regulator/rating, 4
|
|
// aggregator. Omit to let the source text be classified automatically.
|
|
// A Tier-4-only headline metric (revenue/PAT/EBITDA/debt/equity/OCF) fails.
|
|
"source_tier": 1,
|
|
"alt": [
|
|
{"value": 0, "source": "screener.in (cross-check only), dated"}
|
|
]
|
|
}
|
|
],
|
|
"not_available": [
|
|
"any metric you looked for and could not source"
|
|
]
|
|
}
|
|
"""
|
|
|
|
|
|
def example_document() -> Dict[str, Any]:
|
|
"""Return a filled example that deliberately trips a few warnings.
|
|
|
|
It is intentionally clean of errors so --example demonstrates the
|
|
PASS-WITH-WARNINGS path: a ~3% cross-source gap, a datapoint missing its
|
|
unit, a gathered set that stops a quarter behind the latest reported period,
|
|
and a non-headline ratio sourced only from an aggregator -- the common intake
|
|
defects the review found. Every headline metric traces to a filing, so the
|
|
tier check raises no error.
|
|
"""
|
|
return {
|
|
"company": "Meridian Bank Ltd",
|
|
"ticker": "NSE:MERIDBANK",
|
|
"as_of": "2026-07-24",
|
|
"reporting_basis": "consolidated",
|
|
"currency": "INR",
|
|
"units": "crore",
|
|
"fiscal_year_end": "March",
|
|
"latest_reported_period": {"period": "Q1 FY27", "published": "2026-07-11"},
|
|
"datapoints": [
|
|
{"metric": "net_interest_income", "value": 18200, "period": "FY26",
|
|
"basis": "consolidated", "unit": "INR crore", "source_tier": 1,
|
|
"source": "FY26 annual report, p.118",
|
|
"alt": [{"value": 18150, "source": "screener.in 2026-07-20"}]},
|
|
{"metric": "net_profit", "value": 8200, "period": "FY26",
|
|
"basis": "consolidated", "unit": "INR crore", "source_tier": 1,
|
|
"source": "FY26 annual report, p.104",
|
|
"alt": [{"value": 8460, "source": "screener.in 2026-07-20"}]},
|
|
{"metric": "gross_npa_pct", "value": 1.9, "period": "FY26",
|
|
"basis": "consolidated",
|
|
"source": "FY26 annual report, asset-quality note"},
|
|
{"metric": "gross_npa_pct", "value": 3.1, "period": "FY26",
|
|
"entity": "Rival Bank Ltd", "basis": "consolidated", "unit": "%",
|
|
"source": "Rival Bank FY26 results filing, 2026-05-08"},
|
|
{"metric": "casa_ratio_pct", "value": 44.0, "period": "FY26",
|
|
"basis": "consolidated", "unit": "%",
|
|
"source": "FY26 investor presentation, slide 12"},
|
|
{"metric": "pe_ratio", "value": 14.2, "period": "FY26",
|
|
"basis": "consolidated", "unit": "x",
|
|
"source": "screener.in 2026-07-20"},
|
|
],
|
|
"not_available": ["provision_coverage_ratio"],
|
|
}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# CLI
|
|
# --------------------------------------------------------------------------- #
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
"""Construct the command-line argument parser."""
|
|
parser = argparse.ArgumentParser(
|
|
prog=SCRIPT,
|
|
description="Verify gathered stock-analysis data BEFORE analysing it: "
|
|
"provenance, cross-source agreement, basis/unit consistency, "
|
|
"period alignment and recency.",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
)
|
|
parser.add_argument("input", nargs="?", help="path to the intake JSON file")
|
|
parser.add_argument("--json", dest="as_json", action="store_true",
|
|
help="emit findings as JSON instead of a text report")
|
|
parser.add_argument("--template", action="store_true",
|
|
help="print an annotated blank intake file and exit")
|
|
parser.add_argument("--example", action="store_true",
|
|
help="run the built-in example (surfaces sample warnings)")
|
|
return parser
|
|
|
|
|
|
def main(argv: Optional[Sequence[str]] = None) -> int:
|
|
"""Entry point. Returns a process exit code (0 pass, 1 fail, 2 bad input)."""
|
|
parser = build_parser()
|
|
args = parser.parse_args(argv)
|
|
|
|
if args.template:
|
|
sys.stdout.write(TEMPLATE)
|
|
return 0
|
|
|
|
if args.example:
|
|
doc = example_document()
|
|
else:
|
|
if not args.input:
|
|
parser.error("an intake JSON file is required (or use --template / --example)")
|
|
try:
|
|
doc = load_intake(args.input)
|
|
except ValueError as err:
|
|
sys.stderr.write("%s: %s\n" % (SCRIPT, err))
|
|
return 2
|
|
|
|
findings = verify(doc)
|
|
if args.as_json:
|
|
sys.stdout.write(render_json(doc, findings) + "\n")
|
|
else:
|
|
sys.stdout.write(render_text(doc, findings) + "\n")
|
|
|
|
return 1 if counts(findings)[SEVERITY_ERROR] else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|