File size: 8,680 Bytes
d4f8959 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 | # 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}")
|