| |
| """ |
| 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 |
|
|
| |
| |
| |
| 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, |
| }, |
| }, |
| |
| "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, |
| }, |
| }, |
| |
| |
| "Infosys": { |
| "2023": {"revenue": {"value": 1_500_000_000_000, "confidence": "high"}}, |
| "2024": { |
| "revenue": {"value": 1_490_000_000_000, "confidence": "high"}, |
| "net_income": {"value": -50_000_000, "confidence": "high"}, |
| "attrition": 18.0, |
| }, |
| }, |
| |
| "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, |
| }, |
| }, |
| |
| "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}") |
|
|
