#!/usr/bin/env python3 """ End-to-end smoke test of the full calculation pipeline using DemoDataAdapter — no database, no FastAPI, no network. Run with: PYTHONPATH=. python3 scripts/demo_pipeline_smoke_test.py [TICKER ...] For each ticker: DemoDataAdapter -> engines.normalize.build_snapshot -> 30-metric engine -> industry-relative scoring (peer group = the other demo companies in the same sector) -> valuation (WACC/DCF/multiples/blend) -> margin of safety / price bands -> expected return -> Buy/Sell recommendation engine — then prints a Company-Page-shaped summary (spec §36/§55). This is the single script that demonstrates spec §84's "30 metrics се изчисляват... Valuation работи... Fair Value работи... Buy Zone работи... Recommendation работи" end to end without requiring Postgres/Docker — see README.md's build-status note for what this does and doesn't substitute for. """ from __future__ import annotations import sys from datetime import date from app.adapters.demo import DEMO_SEED_PROFILES, DemoDataAdapter from app.engines.metrics import compute_all_metrics from app.engines.normalize import build_snapshot from app.engines.recommendation import SellTriggerInputs, compute_recommendation from app.engines.scoring import ( compute_competitive_advantage_score, compute_financial_health_score, compute_growth_score, compute_overall_score, compute_quality_score, compute_valuation_score, ) from app.engines.scoring.persistence import build_competitive_advantage_proxies from app.engines.valuation import ( CompanyFundamentalsPerShare, DCFAssumptions, ReferenceMultiples, ReferenceSource, WACCInputs, blend_fair_values, cap_terminal_growth_at_risk_free, compute_expected_return, compute_price_bands, compute_wacc, detect_business_profile, run_all_scenarios, ) from app.engines.valuation.expected_return import ExpectedReturnComponents from app.engines.valuation.multiples import compute_multiples_fair_values RISK_FREE_RATE = 0.045 EQUITY_RISK_PREMIUM = 0.045 adapter = DemoDataAdapter() def _snapshot_for(ticker: str): profile = adapter.get_company_profile(ticker) periods = adapter.get_income_statements(ticker) prices = adapter.get_prices(ticker, date.today().replace(day=1), date.today()) current_price = prices[-1].close if prices else None estimates = adapter.get_estimates(ticker) forward_eps = next((e.consensus_value for e in estimates if e.metric == "eps"), None) snap = build_snapshot( security_id=ticker, industry_id=profile.industry, sector_id=profile.sector, periods_most_recent_first=periods, current_price=current_price, forward_eps_estimate=forward_eps, ) return profile, snap def _peer_metric_universe(sector_code: str, metric_keys: list[str]) -> dict[str, dict[str, list[float]]]: """Compute the same metrics for every demo company in the same sector, to build a real (if small) peer group for percentile scoring — same code path a real industry-wide query would feed, just sourced from the demo universe instead of Postgres.""" peers = {k: {"industry_bucket": [], "industry": [], "sector": []} for k in metric_keys} for p in DEMO_SEED_PROFILES: if p.sector_code != sector_code: continue try: _, snap = _snapshot_for(p.ticker) except Exception: continue results = compute_all_metrics(snap) for k in metric_keys: r = results.get(k) if r and r.is_meaningful: peers[k]["sector"].append(r.value) peers[k]["industry"].append(r.value) return peers def run_for_ticker(ticker: str) -> None: profile, snap = _snapshot_for(ticker) c = snap.current # --- WACC --- wacc_inputs = WACCInputs( risk_free_rate=RISK_FREE_RATE, beta=c.beta or profile.beta, equity_risk_premium=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=0.21, market_cap=c.market_cap, total_debt=c.total_debt, ) wacc_result = compute_wacc(wacc_inputs) # --- 30 metrics --- metrics = compute_all_metrics(snap, wacc=wacc_result.value) # --- Scoring (small demo peer group, same sector) --- peer_keys = ["roic", "roic_minus_wacc", "gross_margin", "operating_margin", "net_margin", "fcf_margin", "fcf_growth_cagr_3y", "revenue_growth_cagr_3y", "eps_growth_cagr_3y", "debt_to_ebitda", "net_debt_to_ebitda", "debt_to_equity", "interest_coverage", "current_ratio", "fcf", "revenue_cagr_5y", "pe", "forward_pe", "peg", "ev_to_ebitda", "p_fcf", "ev_to_fcf", "fcf_yield"] peers = _peer_metric_universe(profile.sector, peer_keys) 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) roic_hist = [metrics.get("roic").value] if metrics.get("roic") else [] ca_proxies = build_competitive_advantage_proxies( roic_history=roic_hist * 4, # demo: not enough real trailing history wired up in this smoke test gross_margin_history=[metrics["gross_margin"].value] * 4 if metrics.get("gross_margin") and metrics["gross_margin"].value else [], operating_margin_history=[metrics["operating_margin"].value] * 4 if metrics.get("operating_margin") and metrics["operating_margin"].value else [], revenue_history=[h.revenue for h in ([c] + snap.history)[:5]], fcf_history=[None] * 4, invested_capital_to_revenue=None, ) competitive_advantage = compute_competitive_advantage_score(ca_proxies) scorecard = compute_overall_score(quality, fin_health, growth, competitive_advantage, valuation_score) # --- Valuation: DCF --- dcf_fair_value = None if wacc_result.value and c.revenue and metrics.get("operating_margin") and metrics["operating_margin"].value: terminal_growth = cap_terminal_growth_at_risk_free(0.025, RISK_FREE_RATE) base_assumptions = DCFAssumptions( starting_revenue=c.revenue, revenue_growth_rate=max(min((metrics.get("revenue_growth_cagr_3y").value or 0.03), 0.30), -0.10) if metrics.get("revenue_growth_cagr_3y") and metrics["revenue_growth_cagr_3y"].value else 0.04, operating_margin=metrics["operating_margin"].value, tax_rate=0.21, reinvestment_rate=(c.capital_expenditure / c.revenue) if c.capital_expenditure else 0.05, wacc=wacc_result.value, terminal_growth=terminal_growth, diluted_shares=c.diluted_shares or c.shares_outstanding or 1, net_debt=(c.total_debt or 0) - (c.cash_and_equivalents or 0), ) scenarios = run_all_scenarios(base_assumptions) dcf_fair_value = {k.value: v.fair_value_per_share for k, v in scenarios.items()} # --- Valuation: multiples (using this company's own current multiples as a crude "reference" # stand-in in this single-company smoke test — a real run uses industry/peer medians) --- fundamentals = CompanyFundamentalsPerShare( eps=c.eps_diluted, forward_eps=snap.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, net_debt_per_share=((c.total_debt or 0) - (c.cash_and_equivalents or 0)) / c.diluted_shares if c.diluted_shares else None, ) refs = ReferenceMultiples(pe=18.0, forward_pe=16.0, ev_to_ebitda=11.0, p_fcf=20.0, ev_to_fcf=18.0, source=ReferenceSource.INDUSTRY_MEDIAN) multiples_results = {r.metric: r.fair_value_per_share for r in compute_multiples_fair_values(fundamentals, refs)} profile_type = detect_business_profile(profile.sector, [None]) blended = blend_fair_values( dcf_bear=dcf_fair_value["BEAR"] if dcf_fair_value else None, dcf_base=dcf_fair_value["BASE"] if dcf_fair_value else None, dcf_bull=dcf_fair_value["BULL"] if dcf_fair_value else None, multiples_fair_values=multiples_results, profile=profile_type, ) 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) 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, )) 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=SellTriggerInputs(), ) print(f"\n{'=' * 70}\n{profile.legal_name} [{ticker}] ({profile.country_iso2}, {profile.sector}) — DEMO DATA\n{'=' * 70}") print(f"Price: {c.price:.2f} | Fair Value: {bands.fair_value and round(bands.fair_value, 2)} | " f"MoS: {bands.margin_of_safety and f'{bands.margin_of_safety:.1%}'} | Expected 5Y CAGR: {er.cagr and f'{er.cagr:.1%}'}") print(f"Overall Score: {scorecard.overall and round(scorecard.overall, 1)} | Recommendation: {rec.recommendation.name}") print(f" Quality={quality.value and round(quality.value,1)} FinHealth={fin_health.value and round(fin_health.value,1)} " f"Growth={growth.value and round(growth.value,1)} CompAdv={competitive_advantage.value and (round(competitive_advantage.value,1) if competitive_advantage.value else None)} " f"Valuation={valuation_score.value and round(valuation_score.value,1)}") print(f"WHY: ROIC={metrics['roic'].display} ROIC-WACC={metrics['roic_minus_wacc'].display} " f"FCF Margin={metrics['fcf_margin'].display} Net Debt/EBITDA={metrics['net_debt_to_ebitda'].display} " f"Rev CAGR 5Y={metrics['revenue_cagr_5y'].display}") print(f"RISKS: {rec.risks if rec.risks else 'none triggered'}") print(f"Strong Buy <= {bands.strong_buy_price and round(bands.strong_buy_price,2)} | " f"Buy <= {bands.buy_price and round(bands.buy_price,2)} | " f"Overvalued > {bands.overvalued_price and round(bands.overvalued_price,2)}") print(f"Fair Value confidence: {blended.confidence:.0f}/100 (DCF weight {blended.dcf_weight:.0%}, " f"multiples weight {blended.multiples_weight:.0%}, profile={profile_type.value})") if __name__ == "__main__": tickers = sys.argv[1:] or ["AAPL", "JPM", "NOVO-B", "005930", "4SBK"] for t in tickers: run_for_ticker(t) print(f"\n{len(tickers)} ticker(s) ran through the full pipeline successfully.")