""" Recompute pipeline: FinancialPeriod rows (DB) -> FinancialSnapshot -> metrics -> industry percentiles -> scores -> valuation -> recommendation -> written back to `metrics`, `metric_history`, `valuation`, `scores` (docs/ARCHITECTURE.md §4-8). Triggered by ingestion (app/workers/ingest.py) — never computed inline in an API request (spec §70). """ from __future__ import annotations from datetime import date from sqlalchemy import select from sqlalchemy.orm import Session from app.core.config import get_settings from app.core.db import SessionLocal from app.core.logging import get_logger from app.core.model_version import model_stamp from app.engines.estimates import EstimateRecord, select_forward_estimate from app.engines.metrics import compute_all_metrics from app.engines.metrics.core import invested_capital from app.engines.metrics.formula_utils import safe_div from app.engines.recommendation import build_sell_trigger_inputs, compute_recommendation from app.engines.scoring import ( average_ignoring_none, compute_competitive_advantage_score, compute_confidence_score, compute_data_quality_score, compute_financial_health_score, compute_growth_score, compute_overall_score, compute_quality_score, compute_valuation_score, metric_completeness_inputs, pillar_completeness, years_of_history, ) from app.engines.scoring.overall import MAX_MISSING_WEIGHT_FOR_RENORMALIZATION from app.engines.scoring.persistence import build_competitive_advantage_proxies from app.engines.types import ( ANNUAL_PERIODS_FOR_SNAPSHOT, FinancialSnapshot, LineItems, market_cap_bucket, ) from app.engines.valuation import ( CompanyFundamentalsPerShare, DCFAssumptions, ReferenceMultiples, WACCInputs, blend_fair_values, cap_terminal_growth_at_risk_free, compute_expected_return, compute_price_bands, compute_self_historical_reference_multiples, compute_wacc, detect_business_profile, run_all_scenarios, ) from app.engines.valuation.dcf import EXPLICIT_YEARS, fade_path from app.engines.valuation.expected_return import ExpectedReturnComponents from app.engines.valuation.multiples import combine_reference_multiples, compute_multiples_fair_values from app.models import ( Estimate, FinancialPeriod, Metric, MetricHistory, Price, Score, Security, Source, Valuation, ) from app.workers.celery_app import celery_app logger = get_logger(__name__) def _line_items_from_period(fp: FinancialPeriod, price: float | None) -> LineItems: income = fp.income_statement balance = fp.balance_sheet cash_flow = fp.cash_flow shares_row = fp.shares shares = (shares_row.diluted_shares if shares_row else None) or (shares_row.shares_outstanding if shares_row else None) market_cap = price * shares if (price is not None and shares) else None return LineItems( security_id=fp.security_id, period_end=fp.period_end, period_type=fp.period_type, filing_date=fp.filing_date, currency=fp.currency, revenue=income.revenue if income else None, cogs=income.cogs if income else None, gross_profit=income.gross_profit if income else None, operating_income=income.operating_income if income else None, ebit=income.ebit or (income.operating_income if income else None), ebitda=income.ebitda if income else None, net_income=income.net_income if income else None, eps_diluted=income.eps_diluted if income else None, tax_expense=income.tax_expense if income else None, pretax_income=income.pretax_income if income else None, interest_expense=income.interest_expense if income else None, cash_and_equivalents=balance.cash_and_equivalents if balance else None, short_term_investments=balance.short_term_investments if balance else None, total_debt=balance.total_debt if balance else None, current_assets=balance.current_assets if balance else None, current_liabilities=balance.current_liabilities if balance else None, shareholders_equity=balance.shareholders_equity if balance else None, minority_interest=balance.minority_interest if balance else None, preferred_equity=balance.preferred_equity if balance else None, shares_outstanding=shares_row.shares_outstanding if shares_row else None, diluted_shares=shares_row.diluted_shares if shares_row else None, operating_cash_flow=cash_flow.operating_cash_flow if cash_flow else None, capital_expenditure=cash_flow.capital_expenditure if cash_flow else None, dividends_paid=cash_flow.dividends_paid if cash_flow else None, buybacks=cash_flow.buybacks if cash_flow else None, stock_issuance=cash_flow.stock_issuance if cash_flow else None, price=price, market_cap=market_cap, ) def _build_ttm_current(db: Session, security: Security, as_of: date, price: float | None) -> LineItems | None: """Build a TTM `LineItems` for `security` from its four most recent quarterly periods. Returns None — and logs the reason — when a usable four-quarter window does not exist, so the caller falls back to the annual basis. A silent fallback is exactly the failure mode Part A9 exists to remove: "we are showing you annual data labelled TTM" must be visible in the logs. """ from app.engines.ttm import build_ttm_line_items quarters = ( db.execute( select(FinancialPeriod) .where( FinancialPeriod.security_id == security.id, FinancialPeriod.period_type.in_(("Q1", "Q2", "Q3", "Q4")), FinancialPeriod.period_end <= as_of, ) .order_by(FinancialPeriod.period_end.desc()) .limit(8) ).scalars().all() ) if not quarters: logger.info("recompute.ttm.no_quarterly_periods", security_id=security.id) return None rows = [_line_items_from_period(fp, price if i == 0 else None) for i, fp in enumerate(quarters)] result = build_ttm_line_items(rows) if not result.ok: logger.info("recompute.ttm.unavailable", security_id=security.id, reason=result.reason, quarters_used=result.quarters_used) return None logger.info("recompute.ttm.built", security_id=security.id, period_end=str(result.line_items.period_end)) return result.line_items def build_snapshot_from_db(db: Session, security: Security, as_of: date) -> FinancialSnapshot | None: periods = ( db.execute( select(FinancialPeriod) .where(FinancialPeriod.security_id == security.id, FinancialPeriod.period_type == "FY") .order_by(FinancialPeriod.period_end.desc()) .limit(ANNUAL_PERIODS_FOR_SNAPSHOT) ).scalars().all() ) if not periods: return None latest_price_row = ( db.execute(select(Price).where(Price.security_id == security.id).order_by(Price.date.desc()).limit(1)) .scalar_one_or_none() ) price = latest_price_row.close if latest_price_row else None line_items = [_line_items_from_period(fp, price if i == 0 else None) for i, fp in enumerate(periods)] current, history = line_items[0], line_items[1:] market_cap = current.market_cap # AUDIT FIX (StockLab final engineering pass, Part A9 — docs/AUDIT_TTM_A9.md). # Opt-in real trailing-twelve-month basis for the CURRENT period. Off unless a deployment sets # FINANCIAL_BASIS="TTM" (and has quarterly rows to aggregate, via INGEST_QUARTERLY_PERIODS). # `history` deliberately stays annual: year-over-year growth and 3/5/10-year CAGRs are defined # against comparable annual periods, and mixing a TTM current against annual priors is the # standard, well-understood convention. Mixing TTM into the history rows as well would double # count quarters across overlapping windows. if get_settings().FINANCIAL_BASIS == "TTM": ttm_current = _build_ttm_current(db, security, as_of, price) if ttm_current is not None: current = ttm_current # AUDIT FIX (v4, real-PostgreSQL finding). `market_cap = current.market_cap` used to sit # INSIDE the `if FINANCIAL_BASIS == "TTM"` block above — one indentation level too deep. On the # default configuration (`FINANCIAL_BASIS` is not "TTM") the name was therefore never bound, # and the very next line raised # # UnboundLocalError: local variable 'market_cap' referenced before assignment # # on EVERY recompute. It is read from `current` either way: the TTM branch's only job is to # replace `current`, not to decide whether a market cap exists. # # Why no test caught it: this function needs a `Session`, and `sqlalchemy` cannot be installed # in the build environment (PyPI is blocked by egress policy), so nothing here had ever been # executed. The bucket arithmetic is now `market_cap_bucket()` — pure, and genuinely executed # by tests/test_recompute_snapshot.py — and the real execution path is covered by # tests/integration/test_postgres_pipeline.py. market_cap = current.market_cap bucket = market_cap_bucket(market_cap) # AUDIT FIX (final master pass, §21). `forward_eps_estimate` was never passed here, so it was # None on every real run and `forward_pe` / `eps_growth_forward` were permanently null — # `forward_pe` being a member of the Valuation pillar, every Valuation score was quietly # computed from six of its seven metrics. The `estimates` table is now written by # `app/workers/ingest.py::_ingest_estimates()`, and `select_forward_estimate()` applies §21's # three rules to what is stored: no consensus published after `as_of` (look-ahead), no period # that has already ended (not a forecast), nothing staler than a year. When nothing qualifies # the answer stays None WITH a logged reason — never an invented forward EPS. forward = select_forward_estimate( [ EstimateRecord( metric=row.metric, period_end=row.period_end, consensus_value=row.consensus_value, as_of_date=row.as_of_date, num_analysts=row.num_analysts, source=None, ) for row in db.execute( select(Estimate).where( Estimate.security_id == security.id, Estimate.metric == "eps", Estimate.as_of_date <= as_of, ) ).scalars().all() ], as_of=as_of, ) if forward.value is None: logger.info("recompute.forward_eps.unavailable", security_id=security.id, reason=forward.reason, candidates=forward.candidates_considered) return FinancialSnapshot( security_id=security.id, industry_id=security.company.industry_id, sector_id=security.company.sector.code, market_cap_bucket=bucket, calculation_date=as_of, current=current, history=history, forward_eps_estimate=forward.value, ) def _write_metrics(db: Session, security_id: str, industry_id: str, metrics: dict, as_of: date) -> None: for key, result in metrics.items(): existing = db.query(Metric).filter_by(security_id=security_id, metric_key=key).one_or_none() if existing is None: existing = Metric(security_id=security_id, metric_key=key, industry_id=industry_id) db.add(existing) existing.value = result.value existing.status = result.status.value existing.applicability = result.applicability.value existing.formula_version = result.formula_version existing.as_of = as_of existing.inputs_used = result.inputs_used existing.note = result.note existing.industry_id = industry_id # AUDIT FIX (v4, real-PostgreSQL finding). This used to be an unconditional # `db.add(MetricHistory(...))`, so a SECOND recompute for the same security and date died # with # # psycopg.errors.UniqueViolation: duplicate key value violates unique constraint # "uq_metric_history" DETAIL: Key (security_id, metric_key, as_of, formula_version) # # and took the whole transaction with it — including the Metric, Score and Valuation rows # written around it. Recompute is meant to be re-runnable (a nightly job, a manual # refresh, a retry after a transient failure); it was in fact single-shot per day. # # The snapshot is now upserted on the constraint's own key. This is deliberately NOT a # relaxation of the constraint: a snapshot is "this metric, for this security, as of this # date, under this formula version", and re-deriving it must overwrite that one row rather # than accumulate near-duplicates that later break `_sell_trigger_inputs_from_history()`'s # per-date series. A new `as_of` still creates a new snapshot, and a new `formula_version` # still creates a separate historical version — both are part of the key. snapshot = db.query(MetricHistory).filter_by( security_id=security_id, metric_key=key, as_of=as_of, formula_version=result.formula_version, ).one_or_none() if snapshot is None: snapshot = MetricHistory( security_id=security_id, metric_key=key, as_of=as_of, formula_version=result.formula_version, ) db.add(snapshot) snapshot.value = result.value snapshot.status = result.status.value snapshot.applicability = result.applicability.value snapshot.calculation_date = as_of _REFERENCE_MULTIPLE_METRIC_KEYS = ("pe", "forward_pe", "ev_to_ebitda", "p_fcf", "ev_to_fcf") _SELL_TRIGGER_METRIC_KEYS = ( "roic", "operating_margin", "fcf", "revenue", "eps_diluted", "net_debt_to_ebitda", "interest_coverage", "diluted_shares", ) def _sell_trigger_inputs_from_history(db: Session, security_id: str, as_of: date, price, overvalued_price): """AUDIT FIX (StockLab overhaul, Part 17): real replacement for the previously-empty `SellTriggerInputs()` call — see sell_trigger_builder.py's module docstring for the annual- vs-quarterly caveat this inherits from the TTM finding in AUDIT_METRICS.md.""" rows = db.execute( select(MetricHistory.metric_key, MetricHistory.as_of, MetricHistory.value) .where( MetricHistory.security_id == security_id, MetricHistory.metric_key.in_(_SELL_TRIGGER_METRIC_KEYS), MetricHistory.as_of <= as_of, ) .order_by(MetricHistory.as_of.desc()) .limit(400) ).all() series: dict[str, list] = {k: [] for k in _SELL_TRIGGER_METRIC_KEYS} for metric_key, _as_of, value in rows: if metric_key in series: series[metric_key].append(value) # AUDIT NOTE: dividend_event's CUT/FREEZE/INCREASE label is carried in MetricResult.note # (core.py::dividend_growth), but MetricHistory has no `note` column to persist it (found # while wiring this up — see docs/AUDIT_METRICS.md and docs/AUDIT_ACCOUNTING_QUALITY_A8.md for the # MetricHistory-lacks-note gap generally). Rather than adding a migration mid-audit for a # single field, this reads the already-persisted `dividend_growth_yoy` numeric value instead — # a YoY decline in dividends-per-share is the same underlying signal a CUT event encodes, # recoverable without a schema change. dgy_rows = db.execute( select(MetricHistory.value) .where(MetricHistory.security_id == security_id, MetricHistory.metric_key == "dividend_growth_yoy", MetricHistory.as_of <= as_of) .order_by(MetricHistory.as_of.desc()) .limit(1) ).all() dividend_cut_history = ["DIVIDEND_CUT" if (r[0] is not None and r[0] < 0) else None for r in dgy_rows] return build_sell_trigger_inputs( price=price, overvalued_price=overvalued_price, roic_history=series["roic"], operating_margin_history=series["operating_margin"], fcf_history=series["fcf"], revenue_history=series["revenue"], eps_history=series["eps_diluted"], net_debt_to_ebitda_history=series["net_debt_to_ebitda"], interest_coverage_history=series["interest_coverage"], diluted_shares_history=series["diluted_shares"], dividend_event_history=dividend_cut_history, ) def _historical_reference_multiples(db: Session, security_id: str, as_of: date) -> ReferenceMultiples: """ AUDIT (StockLab overhaul, Part 12): real replacement for the hardcoded ReferenceMultiples(pe=18.0, ...) placeholder that previously stood in for "industry median" regardless of what industry-median actually was. Pulls the security's OWN historical metric values (never another company's) from `metric_history`, excluding today's just-computed row, and takes the median per multiple — see engines/valuation/multiples.py::compute_self_historical_reference_multiples for the "fewer than 3 real points -> None, not a guess" rule. Peer/Industry medians still require a cross-security aggregate query this pass doesn't build (same `peer_metric_values`-supplied-by- caller dependency the scoring engine already documents) — so this function only ever returns SELF_HISTORICAL_5Y_MEDIAN, with individual fields None (NOT AVAILABLE) when history is thin. """ rows = db.execute( select(MetricHistory.metric_key, MetricHistory.value) .where( MetricHistory.security_id == security_id, MetricHistory.metric_key.in_(_REFERENCE_MULTIPLE_METRIC_KEYS), MetricHistory.as_of < as_of, MetricHistory.value.is_not(None), ) .order_by(MetricHistory.as_of.desc()) .limit(200) ).all() historical: dict[str, list[float]] = {k: [] for k in _REFERENCE_MULTIPLE_METRIC_KEYS} for metric_key, value in rows: if metric_key in historical: historical[metric_key].append(value) return compute_self_historical_reference_multiples(historical) def _source_tiers_for_security(db: Session, security_id: str, as_of: date, limit: int = 20) -> list[str]: """AUDIT (StockLab overhaul, Part A1): real input for `compute_data_quality_score()`'s `source_tiers` parameter — the provider source-tier hierarchy (OFFICIAL_FILING > PRIMARY > SECONDARY > CALCULATED, docs/DATA_SOURCES.md §4) behind the `FinancialPeriod` rows that fed this security's current snapshot. One batched query joining `financial_periods.source_id` to `sources.provider_tier` (never a per-period query in a loop — see docs/AUDIT_PERFORMANCE.md's N+1 findings for why that discipline matters here too), bounded to the most recent `limit` periods so a security with a very long history doesn't pull its entire filing record just to characterize present-day data quality. Periods with no `source_id` (possible for demo/synthetic data — `source_id` is nullable, see app/models/financials.py) are simply excluded rather than guessed at; `compute_data_quality_score()` itself already handles an empty/partial `source_tiers` list without fabricating a score for missing entries.""" rows = db.execute( select(Source.provider_tier) .select_from(FinancialPeriod) .join(Source, FinancialPeriod.source_id == Source.id) .where(FinancialPeriod.security_id == security_id, FinancialPeriod.period_end <= as_of) .order_by(FinancialPeriod.period_end.desc()) .limit(limit) ).all() return [tier for (tier,) in rows] def recompute_security(db: Session, security_id: str, peer_metric_values: dict | None = None, industry_medians: dict | None = None, as_of: date | None = None) -> None: """`peer_metric_values` (industry-percentile universe) is supplied by the caller — computing it requires a cross-security aggregate query, kept out of this function to keep it unit-testable with a hand-built peer set; see app/workers/peer_groups.py (IMPLEMENTED in Part B2 — see SPEC_COVERAGE.md) for the production aggregate-query version.""" settings = get_settings() security = db.get(Security, security_id) if security is None: logger.warning("recompute.security_not_found", security_id=security_id) return # `as_of` defaults to today, which is what every production caller wants. It is a parameter # (v4) so that a backfill, or a test proving that a NEW as_of creates a NEW MetricHistory # snapshot, can name the date instead of having to move the system clock. as_of = as_of or date.today() snapshot = build_snapshot_from_db(db, security, as_of) if snapshot is None: logger.warning("recompute.no_financial_data", security_id=security_id) return c = snapshot.current metrics = compute_all_metrics(snapshot) # WACC needs a tax rate that itself comes from roic()'s # effective/fallback tax-rate logic below — computed once, metrics recomputed with WACC after. # AUDIT (StockLab overhaul, Part 9/11): tax_rate was hardcoded to 0.21 here regardless of the # metrics engine's own effective_tax_rate/_fallback_tax_rate logic (core.py::roic) — the two # could silently disagree. Now reuses roic()'s already-computed rate (ACTUAL/3Y-average when # available) and only falls back to the configurable default when roic() itself had nothing to # fall back on (e.g. missing pretax income entirely). roic_tax_rate = metrics["roic"].inputs_used.get("tax_rate") if metrics.get("roic") else None tax_rate = roic_tax_rate if roic_tax_rate is not None else settings.DEFAULT_CORPORATE_TAX_RATE wacc_result = compute_wacc( WACCInputs( risk_free_rate=settings.DEFAULT_RISK_FREE_RATE, beta=c.beta, equity_risk_premium=settings.DEFAULT_EQUITY_RISK_PREMIUM, cost_of_debt_pretax=(c.interest_expense / c.total_debt) if c.interest_expense and c.total_debt else None, tax_rate=tax_rate, market_cap=c.market_cap, total_debt=c.total_debt, ), default_beta_if_missing=settings.DEFAULT_BETA_IF_MISSING, min_cost_of_debt=settings.MIN_COST_OF_DEBT, ) metrics = compute_all_metrics(snapshot, wacc=wacc_result.value) _write_metrics(db, security_id, snapshot.industry_id, metrics, as_of) peers = peer_metric_values or {} quality = compute_quality_score(metrics, peers) fin_health = compute_financial_health_score(metrics, peers) growth = compute_growth_score(metrics, peers) valuation_score = compute_valuation_score(metrics, peers) # AUDIT FIX (final master pass, §8) — the third "engine built, tested, never wired" defect of # the same family as Part A1's Confidence/Data Quality and Part B2's peer groups. # This line used to read `compute_competitive_advantage_score({})` — an EMPTY proxy dict, with # a comment saying the multi-year proxy build was still needed. `build_competitive_advantage_ # proxies()` in app/engines/scoring/persistence.py has existed and been unit-tested since the # 0.2.0 overhaul; nothing ever called it. Because the score requires at least # MIN_PROXIES_FOR_COMPETITIVE_ADVANTAGE (3) non-None proxies, an empty dict meant the # Competitive Advantage pillar was **INSUFFICIENT_DATA for every security, always** — 15% of # the Overall Score weight permanently absent, and `compute_overall_score()` quietly # renormalising the other four pillars to cover it. # Every input below comes from the snapshot this function already has. Nothing is invented: # `invested_capital_to_revenue` uses the same `invested_capital()` helper ROIC uses, and # `customer_concentration_pct` stays None because no adapter ingests it (a disclosed-only # datum) — build_competitive_advantage_proxies() drops a None proxy rather than defaulting it. ca_periods = [c] + list(snapshot.history) ic_current = invested_capital(c) competitive_advantage = compute_competitive_advantage_score( build_competitive_advantage_proxies( roic_history=[ safe_div(p.ebit * (1 - tax_rate), invested_capital(p)) if (p.ebit is not None and invested_capital(p)) else None for p in ca_periods ], gross_margin_history=[safe_div(p.gross_profit, p.revenue) for p in ca_periods], operating_margin_history=[safe_div(p.operating_income, p.revenue) for p in ca_periods], revenue_history=[p.revenue for p in ca_periods], fcf_history=[ (p.operating_cash_flow - p.capital_expenditure) if (p.operating_cash_flow is not None and p.capital_expenditure is not None) else None for p in ca_periods ], invested_capital_to_revenue=safe_div(ic_current, c.revenue), customer_concentration_pct=None, # not ingested by any adapter — see DATA_SOURCES.md ) ) scorecard = compute_overall_score(quality, fin_health, growth, competitive_advantage, valuation_score, weights={ "quality": settings.SCORE_WEIGHT_QUALITY, "financial_health": settings.SCORE_WEIGHT_FINANCIAL_HEALTH, "growth": settings.SCORE_WEIGHT_GROWTH, "competitive_advantage": settings.SCORE_WEIGHT_COMPETITIVE_ADVANTAGE, "valuation": settings.SCORE_WEIGHT_VALUATION, }) dcf_fv = {"BEAR": None, "BASE": None, "BULL": None} if wacc_result.value and c.revenue and metrics.get("operating_margin") and metrics["operating_margin"].value is not None: # AUDIT (StockLab overhaul, Part 9/11): the three fallback constants below (revenue growth # 0.04, reinvestment rate 0.05, terminal growth 0.025) were bare literals; now sourced from # Settings so they're visible/overridable rather than buried here. They remain fallbacks — # used only when the security's own computed metric is unavailable — and this DCF's # single-rate-across-10-years architecture (not per-year assumptions) and multiplier-based # Bear/Bull scenarios are audited in docs/AUDIT_VALUATION.md, not changed in this pass. near_term_growth = ( metrics["revenue_growth_cagr_3y"].value if metrics.get("revenue_growth_cagr_3y") and metrics["revenue_growth_cagr_3y"].value else settings.DEFAULT_DCF_REVENUE_GROWTH_FALLBACK ) terminal_growth = cap_terminal_growth_at_risk_free( settings.DEFAULT_DCF_TERMINAL_GROWTH, settings.DEFAULT_RISK_FREE_RATE) # AUDIT FIX (Part B1): the near-term growth rate now fades toward the terminal rate across # the explicit window instead of being held flat for 10 years. Opt-out via # DCF_GROWTH_FADE=false, which restores the pre-B1 flat path exactly (proven equivalent by # test_no_path_supplied_is_identical_to_the_flat_scalar). No margin path is set here: the # platform has no defensible source for a per-company mature-margin target, and inventing # one would be exactly the fabrication this project forbids -- `operating_margin_path` is # available for a caller that does have one. See docs/AUDIT_DCF_B1.md. growth_path = ( fade_path(near_term_growth, terminal_growth, years=EXPLICIT_YEARS, fade_years=settings.DCF_GROWTH_FADE_YEARS) if settings.DCF_GROWTH_FADE else None ) # AUDIT FIX (final master pass, §0 "NO SILENT FALLBACKS" — two real defects on one line). # # 1. `diluted_shares=c.diluted_shares or 1` — CRITICAL. `run_dcf()` opens with # `if a.diluted_shares is None or a.diluted_shares <= 0: return `, precisely so a # missing share count can never produce a fair value. Substituting 1 here defeated that # guard before it could ever run: the whole equity value of the company was divided by # one share. A company with a $2bn equity value and no ingested share count produced a # "fair value per share" of 2,000,000,000 against a $25 price — an effectively infinite # margin of safety and a STRONG BUY, from a company about which the platform did not # know the most basic fact. Now passed through unchanged so the engine's own guard # fires and the DCF is correctly reported as unavailable. # # 2. `net_debt=(c.total_debt or 0) - (c.cash_and_equivalents or 0)` — an UNKNOWN debt # balance was treated as ZERO debt, turning any indebted company whose debt failed to # ingest into a net-cash company: enterprise value understated, equity value and fair # value overstated, in the optimistic direction. Now the DCF is skipped when `total_debt` # is unknown, rather than valuing the company as if it had none. A genuine 0 is # unaffected — a debt-free company reports `total_debt = 0.0`, not None. if c.diluted_shares is None or c.diluted_shares <= 0 or c.total_debt is None: logger.info( "recompute.dcf.skipped_insufficient_inputs", security_id=security_id, diluted_shares=c.diluted_shares, total_debt=c.total_debt, reason=("missing_or_non_positive_share_count" if (c.diluted_shares is None or c.diluted_shares <= 0) else "unknown_total_debt"), ) else: base = DCFAssumptions( starting_revenue=c.revenue, revenue_growth_rate=near_term_growth, revenue_growth_path=growth_path, operating_margin=metrics["operating_margin"].value, tax_rate=tax_rate, reinvestment_rate=(c.capital_expenditure / c.revenue) if c.capital_expenditure and c.revenue else settings.DEFAULT_DCF_REINVESTMENT_RATE_FALLBACK, wacc=wacc_result.value, terminal_growth=terminal_growth, diluted_shares=c.diluted_shares, net_debt=c.total_debt - (c.cash_and_equivalents or 0), ) scenarios = run_all_scenarios(base, mode=settings.DCF_SCENARIO_MODE) dcf_fv = {k.value: v.fair_value_per_share for k, v in scenarios.items()} fundamentals = CompanyFundamentalsPerShare( eps=c.eps_diluted, forward_eps=snapshot.forward_eps_estimate, ebitda_per_share=(c.ebitda / c.diluted_shares) if c.ebitda and c.diluted_shares else None, fcf_per_share=metrics["fcf_per_share"].value if metrics.get("fcf_per_share") else None, # AUDIT FIX (final master pass): same unknown-debt-as-zero-debt fallback as the DCF block # above. `net_debt_per_share` feeds every EV-based multiple's fair value # (ev_to_ebitda, ev_to_fcf), so treating an unknown debt balance as no debt inflates those # fair values. None when debt is unknown -> those multiples correctly return None rather # than an optimistic number. A genuinely debt-free company reports 0.0, not None. net_debt_per_share=( (c.total_debt - (c.cash_and_equivalents or 0)) / c.diluted_shares if (c.total_debt is not None and c.diluted_shares) else None ), ) # AUDIT FIX (StockLab overhaul, Part 12): real self-historical reference multiples (median of # this security's own past pe/forward_pe/ev_to_ebitda/p_fcf/ev_to_fcf from metric_history), # replacing the hardcoded industry-median-shaped placeholder that stood here before this # audit. Peer/Industry medians are still NOT AVAILABLE — see # _historical_reference_multiples()'s docstring and docs/AUDIT_VALUATION.md. # AUDIT FIX (Part B2): the industry median is a real, computed reference now # (app/workers/peer_groups.py::industry_reference_multiples), not an enum value nothing # produced. `industry_medians` is None when the caller did not supply one or the industry had # too few reporters, in which case combine_reference_multiples() degrades to exactly the # pre-B2 self-historical behaviour. Set MULTIPLES_REFERENCE_PREFERENCE="self_only" to keep # the pre-B2 behaviour unconditionally. refs = combine_reference_multiples( _historical_reference_multiples(db, security_id, as_of), industry_medians, preference=settings.MULTIPLES_REFERENCE_PREFERENCE, ) multiples_results = {r.metric: r.fair_value_per_share for r in compute_multiples_fair_values(fundamentals, refs)} profile_type = detect_business_profile(snapshot.sector_id, [h.operating_cash_flow for h in ([c] + snapshot.history)]) # AUDIT FIX (StockLab final engineering pass, Part B3 -- docs/AUDIT_FAIR_VALUE_B3.md). # `blend_fair_values()` takes `data_completeness_pct` and defaults it to 100.0, and this call # site -- the only one in the application -- never passed it. So BlendedFairValue.confidence # started from a perfect 100 for EVERY security no matter how much of its data was missing, # and could then only be reduced by DCF/multiples disagreement and business-profile # complexity. A security with three usable metrics reported the same completeness # contribution as one with all thirty. The real number was already being computed a few dozen # lines below for the Confidence Score; it is now computed once, here, and used by both. meaningful_statuses, real_metric_gaps = metric_completeness_inputs(metrics) _metrics_considered = len(meaningful_statuses) + real_metric_gaps data_completeness_pct = ( 100.0 * len(meaningful_statuses) / _metrics_considered if _metrics_considered else 0.0 ) blended = blend_fair_values( dcf_fv["BEAR"], dcf_fv["BASE"], dcf_fv["BULL"], multiples_results, profile_type, data_completeness_pct=data_completeness_pct, ) risk_score_proxy = fin_health.value if fin_health.value is not None else 50.0 bands = compute_price_bands( blended.weighted_fair_value, c.price, risk_score_0_100=risk_score_proxy, strong_buy_mos=settings.DEFAULT_STRONG_BUY_MOS, buy_mos=settings.DEFAULT_BUY_MOS, overvalued_premium=settings.DEFAULT_OVERVALUED_PREMIUM, ) er = compute_expected_return(ExpectedReturnComponents( fundamental_growth_rate=(metrics["eps_growth_cagr_3y"].value if metrics.get("eps_growth_cagr_3y") and metrics["eps_growth_cagr_3y"].value else None), shareholder_yield=(metrics["shareholder_yield"].value if metrics.get("shareholder_yield") else 0.0), current_multiple=metrics["pe"].value if metrics.get("pe") else None, reference_multiple=refs.pe, years=5, )) # AUDIT FIX (StockLab overhaul, Part 17): real historical-input pipeline replacing the # previously-empty SellTriggerInputs() — see _sell_trigger_inputs_from_history() above. trigger_inputs = _sell_trigger_inputs_from_history(db, security_id, as_of, c.price, bands.overvalued_price) rec = compute_recommendation( overall_score=scorecard.overall, margin_of_safety=bands.margin_of_safety, risk_score_0_100=risk_score_proxy, expected_cagr_base=er.cagr, sell_trigger_inputs=trigger_inputs, ) # AUDIT FIX (StockLab overhaul, Part A1): compute_confidence_score()/compute_data_quality_score() # (app/engines/scoring/{confidence,data_quality}.py) existed with unit tests since an earlier # pass of this overhaul but were never actually called here — docs/AUDIT_GUI.md flagged this as # a real end-to-end wiring gap. Now wired using only data already computed above (pillar # scores, metrics, history depth) plus one batched DB query for source tiers # (_source_tiers_for_security) — no fabricated inputs, and both engines already degrade to # `value=None` (never a fake number) when they don't have enough to work with. pillars_present, pillars_total = pillar_completeness( [quality.value, fin_health.value, growth.value, competitive_advantage.value, valuation_score.value] ) # meaningful_statuses / real_metric_gaps were computed above, before the Fair Value blend # (Part B3) -- reused here rather than recomputed, so the completeness the blend's confidence # is based on and the completeness the Confidence Score is based on can never disagree. confidence = compute_confidence_score( pillars_present=pillars_present, pillars_total=pillars_total, metric_statuses=meaningful_statuses, metrics_missing_count=real_metric_gaps, years_of_history=years_of_history(len(snapshot.history)), dcf_fair_value=dcf_fv["BASE"], multiples_fair_value=average_ignoring_none(list(multiples_results.values())), ) data_quality = compute_data_quality_score( source_tiers=_source_tiers_for_security(db, security_id, as_of), field_statuses=meaningful_statuses, ) val_row = db.query(Valuation).filter_by(security_id=security_id, calculation_date=as_of).one_or_none() if val_row is None: val_row = Valuation(security_id=security_id, calculation_date=as_of) db.add(val_row) val_row.wacc = wacc_result.value val_row.dcf_bear_fair_value, val_row.dcf_base_fair_value, val_row.dcf_bull_fair_value = dcf_fv["BEAR"], dcf_fv["BASE"], dcf_fv["BULL"] val_row.multiples_fair_values = multiples_results val_row.business_profile = profile_type.value val_row.bear_fair_value, val_row.base_fair_value, val_row.bull_fair_value = blended.bear_fair_value, blended.base_fair_value, blended.bull_fair_value val_row.weighted_fair_value = blended.weighted_fair_value val_row.fair_value_confidence = blended.confidence val_row.price_at_calculation = c.price val_row.margin_of_safety = bands.margin_of_safety val_row.strong_buy_price, val_row.buy_price, val_row.overvalued_price = bands.strong_buy_price, bands.buy_price, bands.overvalued_price val_row.expected_return_5y = er.cagr score_row = db.query(Score).filter_by(security_id=security_id, calculation_date=as_of).one_or_none() if score_row is None: score_row = Score(security_id=security_id, calculation_date=as_of) db.add(score_row) score_row.quality_score, score_row.financial_health_score = quality.value, fin_health.value score_row.growth_score, score_row.competitive_advantage_score = growth.value, competitive_advantage.value score_row.valuation_score, score_row.overall_score = valuation_score.value, scorecard.overall score_row.risk_score = risk_score_proxy # AUDIT FIX (StockLab overhaul, Part A1): previously unset (column existed on no prior schema # at all — see alembic/versions/0002_confidence_data_quality_scores.py). None when the engine # itself returned None (e.g. zero metrics computed at all) — never coerced to 0. score_row.confidence_score = confidence.value score_row.confidence_components = ( [{"name": c.name, "weight": c.weight, "raw_score": c.raw_score, "contribution": c.contribution} for c in confidence.components] or None ) score_row.data_quality_score = data_quality.value score_row.data_quality_components = ( [{"name": c.name, "weight": c.weight, "raw_score": c.raw_score, "contribution": c.contribution} for c in data_quality.components] or None ) score_row.weights_used = scorecard.weights_used # AUDIT FIX (final master pass, §9 "score must be explainable" + §40 data lineage). # `scores.subscore_detail` is a JSON column carrying the comment "per spec §49 audit trail". # Nothing ever wrote it and nothing ever read it — the explainability payload the spec # requires existed as an empty column. `compute_overall_score()` has always returned # `missing_weight` and `subscores_included_in_overall`, and each `ScoreResult` has always # carried `metrics_included` / `metrics_excluded` / `peer_group_tier`; none of it reached the # API, so a consumer could see "Overall Score: 88" with no way to learn that it rests on four # of five pillars, which metrics were dropped, or how wide the peer group was. # # `missing_weight` is the number §9 specifically demands be visible: renormalisation must not # hide missing information. It is the fraction of total pillar weight that had no value and # was redistributed over the rest. score_row.subscore_detail = { "missing_weight": scorecard.missing_weight, "subscores_included_in_overall": scorecard.subscores_included_in_overall, "max_missing_weight_allowed": MAX_MISSING_WEIGHT_FOR_RENORMALIZATION, "pillars": { name: { "value": result.value, "status": result.status, "metrics_included": result.metrics_included, "metrics_excluded": result.metrics_excluded, "peer_group_tier": result.peer_group_tier, "weight": scorecard.weights_used.get(name), } for name, result in ( ("quality", quality), ("financial_health", fin_health), ("growth", growth), ("competitive_advantage", competitive_advantage), ("valuation", valuation_score), ) }, **model_stamp(), } score_row.recommendation = rec.recommendation.name score_row.triggers_fired = rec.risks db.commit() logger.info("recompute.security.complete", security_id=security_id, overall_score=scorecard.overall, recommendation=rec.recommendation.name) @celery_app.task(name="app.workers.recompute.recompute_security_task") def recompute_security_task(security_id: str) -> None: """AUDIT FIX (StockLab final engineering pass, Part B2 — docs/AUDIT_PEER_GROUPS_B2.md). This task previously called `recompute_security(db, security_id)` with no peer universe, and it was the ONLY call site in the application — so `peer_metric_values` was `{}` on every real run, `percentile_rank()` was handed an empty peer list for every metric and returned its neutral 50.0, and every security's every pillar score came out at exactly 50. The scoring arithmetic was correct throughout; it was never given anything to compare against. Note the cost shape: this rebuilds the peer universe for ONE security. For a universe-wide refresh use `app.workers.peer_groups.recompute_universe_task`, which loads it once for all. """ from app.workers.peer_groups import ( industry_medians_for, industry_reference_multiples_from_universe, load_metric_universe, peer_metric_values_for, ) settings = get_settings() db = SessionLocal() try: universe = load_metric_universe(db) all_medians = industry_reference_multiples_from_universe( universe, min_group_size=settings.INDUSTRY_MULTIPLE_MIN_GROUP_SIZE, ) recompute_security( db, security_id, peer_metric_values=peer_metric_values_for(universe, security_id), industry_medians=industry_medians_for(all_medians, universe, security_id), ) finally: db.close()