# backend/recommendations.py """ Phase 2 — Recommendation engine. Consumes red_flags.py's output directly. No LLM involvement — scored decision tree only, so this works in Mode C (no LLM available) exactly like red_flags.py does. DECISION LOGIC (stated explicitly so it's auditable, not buried in code): BUY if zero flags triggered at all AVOID if ANY single flag has severity >= HIGH_SEVERITY_THRESHOLD (30), OR total risk_score > RISK_SCORE_AVOID_THRESHOLD (50) — these are two independent triggers, either one is sufficient. A company with one severe problem (e.g. negative net income, severity 35) is AVOID even if that's its only flag and its aggregate score (35) is under the 50 cutoff. A company with several minor flags that sum past 50 is also AVOID, even if no single flag was individually severe. This asymmetry is deliberate — see the conversation that designed it: a single severe problem shouldn't get diluted into a HOLD just because nothing else is wrong. HOLD if flags exist but none are severe and risk_score <= 50 SKIP if the sector isn't one red_flags.py covers with real ratio thresholds (currently BANK, IT, MANUFACTURING only — PHARMA/ ENERGY/GENERAL have weaker derived-only coverage per red_flags.py's own docstring), or if confidence is "low", or if red_flags.py itself returned an error (no filing found, etc.) SKIP is a real, honest answer here — not a fallback for "couldn't be bothered." Calling BUY/HOLD/AVOID on sectors with thin metric coverage would be worse than admitting the data doesn't support a call. """ from backend.red_flags import evaluate_red_flags, FLAG_SEVERITY # Sectors where red_flags.py has real extracted-ratio thresholds (not just # derived proxies or revenue-trend-only checks). Matches the coverage gap # documented in red_flags.py's own module docstring. SUPPORTED_SECTORS = {"BANK", "IT", "MANUFACTURING"} HIGH_SEVERITY_THRESHOLD = 30 RISK_SCORE_AVOID_THRESHOLD = 50 def recommend_from_red_flags(red_flags_result: dict) -> dict: """ Pure function: takes red_flags.py's output dict and returns a recommendation. Separated from evaluate_recommendation() below so it can be tested/reused without a graph instance. """ company = red_flags_result["company"] year = red_flags_result["year"] sector = red_flags_result["sector"] base = { "company": company, "year": year, "sector": sector, "recommendation": None, "risk_score": red_flags_result.get("risk_score"), "confidence": red_flags_result.get("confidence"), "triggers": [], } if red_flags_result.get("error"): base["recommendation"] = "SKIP" base["reason"] = red_flags_result["error"] return base if sector not in SUPPORTED_SECTORS: base["recommendation"] = "SKIP" base["reason"] = ( f"Sector '{sector}' has limited ratio coverage in red_flags.py " f"(derived/proxy metrics only) — not enough signal for a " f"confident recommendation." ) return base confidence = red_flags_result.get("confidence") if confidence == "low": base["recommendation"] = "SKIP" base["reason"] = ( "Overall confidence is low — at least one triggered flag is " "based on low-confidence extracted data, not safe to act on." ) base["triggers"] = [f["message"] for f in red_flags_result["flags_triggered"]] return base flags = red_flags_result["flags_triggered"] risk_score = red_flags_result["risk_score"] if not flags: base["recommendation"] = "BUY" base["reason"] = "No red flags triggered against the extracted metrics." return base high_severity_flags = [ f for f in flags if FLAG_SEVERITY.get(f["flag"], 0) >= HIGH_SEVERITY_THRESHOLD ] if high_severity_flags: base["recommendation"] = "AVOID" worst = high_severity_flags[0]["flag"] base["reason"] = ( f"Single high-severity flag triggered: {worst} " f"(severity {FLAG_SEVERITY.get(worst, 0)} >= {HIGH_SEVERITY_THRESHOLD})" ) base["triggers"] = [f["message"] for f in flags] return base if risk_score > RISK_SCORE_AVOID_THRESHOLD: base["recommendation"] = "AVOID" base["reason"] = ( f"Aggregate risk_score {risk_score} exceeds " f"{RISK_SCORE_AVOID_THRESHOLD}, despite no single severe flag." ) base["triggers"] = [f["message"] for f in flags] return base base["recommendation"] = "HOLD" base["reason"] = ( f"Minor flags present (risk_score {risk_score}), " f"none individually severe." ) base["triggers"] = [f["message"] for f in flags] return base def evaluate_recommendation(graph, company: str, year: str, sector: str = "GENERAL") -> dict: """Main entry point — mirrors evaluate_red_flags()'s signature exactly so main.py can call this the same way.""" red_flags_result = evaluate_red_flags(graph, company, year, sector=sector) return recommend_from_red_flags(red_flags_result) if __name__ == "__main__": class FakeGraph: def __init__(self, data): self._data = data def get_company_metrics(self, company): return self._data.get(company, {}) fake_data = { "HDFC Bank": { "2023": { "deposits": {"value": 1_900_000_00_00_000, "confidence": "high"}, "gross_npa_pct": 1.3, "net_npa_pct": 0.4, "casa_ratio": 44.0, "capital_adequacy": 18.9, }, "2024": { "profit_after_tax": {"value": 608_120_00_00_000, "confidence": "high"}, "deposits": {"value": 1_500_000_00_00_000, "confidence": "high"}, "gross_npa_pct": 6.2, "net_npa_pct": 0.33, "casa_ratio": 28.0, "capital_adequacy": 19.3, }, }, # clean bank, zero flags -> should be BUY "ICICI Bank": { "2023": {"deposits": {"value": 1_000_000_00_00_000, "confidence": "high"}}, "2024": { "profit_after_tax": {"value": 400_000_00_00_000, "confidence": "high"}, "deposits": {"value": 1_050_000_00_00_000, "confidence": "high"}, "gross_npa_pct": 1.1, "net_npa_pct": 0.3, "casa_ratio": 42.0, "capital_adequacy": 17.0, }, }, # one severe flag only (negative net income), nothing else -> # tests the "single severe flag overrides low aggregate score" rule "Infosys": { "2023": {"revenue": {"value": 1_500_000_000_000, "confidence": "high"}}, "2024": { "revenue": {"value": 1_490_000_000_000, "confidence": "high"}, # barely declined, under 5% "net_income": {"value": -50_000_000, "confidence": "high"}, "attrition": 18.0, # under threshold }, }, # CASA only (low severity, 10) -> should be HOLD "AxisBank": { "2023": {"deposits": {"value": 900_000_00_00_000, "confidence": "high"}}, "2024": { "profit_after_tax": {"value": 200_000_00_00_000, "confidence": "high"}, "deposits": {"value": 920_000_00_00_000, "confidence": "high"}, "gross_npa_pct": 2.0, "net_npa_pct": 0.8, "casa_ratio": 25.0, "capital_adequacy": 16.0, }, }, # unsupported sector -> SKIP "SunPharma": { "2024": { "revenue": {"value": 500_000_000_000, "confidence": "high"}, "r_and_d": {"value": 30_000_000_000, "confidence": "high"}, "net_income": {"value": 60_000_000_000, "confidence": "high"}, }, }, } fg = FakeGraph(fake_data) for company, year, sector in [ ("HDFC Bank", "2024", "BANK"), ("ICICI Bank", "2024", "BANK"), ("Infosys", "2024", "IT"), ("AxisBank", "2024", "BANK"), ("SunPharma", "2024", "PHARMA"), ("NoSuchCompany", "2024", "BANK"), ]: result = evaluate_recommendation(fg, company, year, sector=sector) print(f"\n{company} ({sector}, {year}) -> {result['recommendation']}") print(f" reason: {result['reason']}") print(f" risk_score={result['risk_score']} confidence={result['confidence']}") for t in result["triggers"]: print(f" - {t}")