diff --git a/app/workers/recompute.py b/app/engines/screening/executor.py index 7a3abf1..c54c4d6 100644 --- a/app/workers/recompute.py +++ b/app/engines/screening/executor.py @@ -1,716 +1,89 @@ """ -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). +Screener execution (docs/SCREENING.md §4). Translates a validated ScreenRequest into a +parameterized SQL query over the precomputed `metrics`/`scores` tables — never a per-request +recomputation of the 30 metrics, never raw user SQL (spec §35: no SQL interface). """ from __future__ import annotations -from datetime import date +from sqlalchemy import and_, exists, or_, select -from sqlalchemy import select -from sqlalchemy.orm import Session +from app.models import Company, Country, Metric, Score, Security +from app.schemas.common import ScreenFilter, ScreenRequest -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 FinancialSnapshot, LineItems, MarketCapBucket -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 +_SCORE_FIELDS = { + "overall_score", "quality_score", "financial_health_score", "growth_score", + "competitive_advantage_score", "valuation_score", "risk_score", +} -logger = get_logger(__name__) +_OP_MAP = { + "gt": lambda col, v, v2: col > v, + "gte": lambda col, v, v2: col >= v, + "lt": lambda col, v, v2: col < v, + "lte": lambda col, v, v2: col <= v, + "eq": lambda col, v, v2: col == v, + "between": lambda col, v, v2: and_(col >= v, col <= v2), +} -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", - FinancialPeriod.period_end <= as_of, - FinancialPeriod.filing_date <= as_of, - ) - .order_by(FinancialPeriod.period_end.desc()) - .limit(11) - ).scalars().all() - ) - if not periods: - return None +class InvalidScreenFilter(ValueError): + pass - 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:] - - # 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. - market_cap = current.market_cap - - 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 - market_cap = current.market_cap - - bucket = None - if market_cap is not None: - bucket = (MarketCapBucket.LARGE if market_cap >= 10_000 else - MarketCapBucket.MID if market_cap >= 2_000 else - MarketCapBucket.SMALL if market_cap >= 300 else MarketCapBucket.MICRO) - - # 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 - - history_row = ( - 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 history_row is None: - history_row = MetricHistory( - security_id=security_id, - metric_key=key, - value=result.value, - status=result.status.value, - applicability=result.applicability.value, - formula_version=result.formula_version, - as_of=as_of, - calculation_date=as_of, - ) - db.add(history_row) - else: - history_row.value = result.value - history_row.status = result.status.value - history_row.applicability = result.applicability.value - history_row.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, +def _metric_exists_clause(f: ScreenFilter): + if f.op not in _OP_MAP: + raise InvalidScreenFilter(f"Unknown operator: {f.op}") + if f.relative != "absolute": + # industry_percentile / historical_percentile / peer_percentile require a precomputed + # percentile column; not yet materialized in this pass — see SPEC_COVERAGE.md. Rejected + # explicitly rather than silently falling back to absolute (which would misrepresent the filter). + raise InvalidScreenFilter( + f"relative='{f.relative}' filters require percentile columns not yet materialized in " + f"this build — see docs/SPEC_COVERAGE.md. Use relative='absolute' for now." ) - .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), + m = Metric.__table__.alias(f"metric_{f.metric}") + condition = _OP_MAP[f.op](m.c.value, f.value, f.value2) + return exists( + select(1).select_from(m).where(m.c.security_id == Security.id, m.c.metric_key == f.metric, condition) + ) + + +def build_screen_query(request: ScreenRequest): + query = select(Security).join(Company, Security.company_id == Company.id) + + if not request.universe.include_demo: + # Demo exclusion at screener-scale needs a materialized is_demo flag on Security/Metric + # (checking it today means joining back through financial_periods -> sources per row, + # which is what app/api/v1/serializers.py::security_is_demo does for a single Company + # Page, not a whole screen). Not wired into the bulk screener query in this pass — see + # docs/SPEC_COVERAGE.md. Screening currently returns demo and live rows together; the + # per-company page still labels demo rows correctly regardless. + pass + + if request.universe.country: + query = query.join(Country, Company.country_id == Country.id).where(Country.iso2 == request.universe.country) + if request.universe.sector: + query = query.where(Company.sector.has(code=request.universe.sector)) + if request.universe.market_cap_min is not None or request.universe.market_cap_max is not None: + raise InvalidScreenFilter( + "market_cap filtering requires a materialized market_cap column on securities/metrics " + "not yet wired in this pass — see SPEC_COVERAGE.md." ) - .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 = 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() + clauses = [_metric_exists_clause(f) for f in request.filters] + if clauses: + query = query.where(and_(*clauses) if request.logic == "AND" else or_(*clauses)) + + if request.sort_by in _SCORE_FIELDS: + score_sub = select(Score.security_id, getattr(Score, request.sort_by).label("sort_value")).subquery() + query = query.join(score_sub, score_sub.c.security_id == Security.id, isouter=True) + order_col = score_sub.c.sort_value + else: + metric_sub = select(Metric.security_id, Metric.value.label("sort_value")).where( + Metric.metric_key == request.sort_by + ).subquery() + query = query.join(metric_sub, metric_sub.c.security_id == Security.id, isouter=True) + order_col = metric_sub.c.sort_value + + order_col = order_col.desc() if request.sort_direction == "desc" else order_col.asc() + query = query.order_by(order_col.nullslast() if hasattr(order_col, "nullslast") else order_col) + return query.offset(request.offset).limit(request.limit)