""" Strategy cascade recommend — pure, ≥4 paths, no LLM. Feature flags ------------- ENABLE_STRATEGY_CASCADE (default: true) When true (product default), recommend_strategy_paths runs the full 2026 cascade scoring and returns cascade_enforced=true. When false, the same path list is still produced (scores, ≥4 ids, primary) but cascade_enforced=false so callers can treat cascade as advisory/off without losing the path surface. ENABLE_TAX_PATHS (default: true) — see core.strategy.tax_paths When true, PATH_TAX (tax_psi_ulgi) gets tax checklists (PSI / ulga B+R / IP Box) via attach_tax_checklists. When false, PATH_TAX remains in the cascade without checklist payload; tax_paths_enabled=false on the result. """ from __future__ import annotations import os from typing import Any, Dict, List, Optional from core.strategy.direct_eu import attach_direct_eu_meta from core.strategy.tax_paths import attach_tax_checklists, tax_paths_enabled # Canonical path ids (product) PATH_TAX = "tax_psi_ulgi" PATH_REGIONAL_FST = "regional_fst" PATH_NATIONAL_FENG = "national_feng" PATH_DEBT = "debt_loan_guarantee" PATH_DIRECT_EU = "direct_eu_prep" FLAG_STRATEGY_CASCADE = "ENABLE_STRATEGY_CASCADE" def strategy_cascade_enabled() -> bool: """Thin flag helper — default ON so product cascade path stays active.""" return os.environ.get(FLAG_STRATEGY_CASCADE, "true").lower() in ("1", "true", "yes") def recommend_strategy_paths( *, eligibility: Optional[Dict[str, Any]] = None, company_data: Optional[Dict[str, Any]] = None, goal: str = "", region: str = "", ) -> Dict[str, Any]: """ Returns ordered strategy paths with scores and reasons. Always returns at least four distinct path labels among the five canonical ids. cascade_enforced reflects ENABLE_STRATEGY_CASCADE (default true). Paths are always returned; when the flag is off, cascade_enforced is false. """ elig = eligibility or {} cd = company_data or {} dm = elig.get("de_minimis") if isinstance(elig.get("de_minimis"), dict) else {} msp = elig.get("msp") if isinstance(elig.get("msp"), dict) else {} goal_l = (goal or "").lower() region_l = ( region or cd.get("region") or cd.get("voivodeship") or (cd.get("address") or "") ) region_l = str(region_l).lower() fst_regions = ( "śląsk", "slask", "małopol", "malopol", "dolnośląsk", "dolnoslask", "wielkopol", "łódzk", "lodzk", ) in_fst = any(r in region_l for r in fst_regions) paths: List[Dict[str, Any]] = [] # Tax / PSI / ulgi — always relevant for investment-like goals tax_score = 55.0 tax_reasons = ["Ulgi i PSI często pewniejsze niż konkurs grantowy (2026)."] if any(k in goal_l for k in ("inwestyc", "hala", "linia", "maszyn", "cit", "pit", "podat")): tax_score += 20 tax_reasons.append("Cel inwestycyjny — sprawdź PSI / ulgę B+R przed naborem.") if any(k in goal_l for k in ("oprogram", "ip box", "patent", "b+r", "badaw")): tax_score += 15 tax_reasons.append("B+R/IP — ulga B+R i IP Box jako fundament.") # Tax checklists attached after primary marking via attach_tax_checklists batch paths.append( { "id": PATH_TAX, "label": "Podatki / PSI / ulgi (B+R, IP Box)", "score": tax_score, "reasons": tax_reasons, "primary": False, } ) # Regional / FST reg_score = 50.0 reg_reasons = ["Programy regionalne FE — zwykle łatwiejsze wejście dla MŚP."] if in_fst: reg_score += 25 reg_reasons.append("Siedziba w obszarze FST — wyższe % i łagodniejsze kryteria.") if any(k in goal_l for k in ("oze", "termo", "cyfryz", "szkolen", "lokaln")): reg_score += 10 paths.append( { "id": PATH_REGIONAL_FST, "label": "Region / FST (programy wojewódzkie)", "score": reg_score, "reasons": reg_reasons, "primary": False, } ) # National FENG / NCBR grants feng_score = 52.0 feng_reasons = ["FENG/SMART/NCBR — główny trzon krajowych grantów innowacyjnych."] if any(k in goal_l for k in ("b+r", "innowac", "trl", "wdroż", "smart", "deeptech")): feng_score += 22 feng_reasons.append("Cel B+R/innowacja — FENG/SMART lub NCBR.") if msp.get("is_large"): feng_score -= 10 feng_reasons.append("Duża firma — część naborów MŚP niedostępna.") if dm.get("exhausted"): feng_score -= 5 feng_reasons.append("De minimis wyczerpany — unikaj naborów opartych o de minimis.") paths.append( { "id": PATH_NATIONAL_FENG, "label": "Krajowe granty FENG / NCBR / PARP", "score": feng_score, "reasons": feng_reasons, "primary": False, } ) # Debt / guarantee debt_score = 48.0 debt_reasons = ["Instrumenty dłużne BGK/ARP — ciągłe nabory, pomost pod grant."] if dm.get("exhausted"): debt_score -= 25 debt_reasons.append( "Limit de minimis wyczerpany — gwarancje de minimis niedostępne jako primary." ) if any(k in goal_l for k in ("pożyczk", "kredyt", "gwaranc", "płynnoś", "kapitał obrot")): debt_score += 20 if any(k in goal_l for k in ("kpo",)): debt_score += 15 debt_reasons.append("KPO 2026: preferuj instrumenty dłużne zamiast nowych grantów KPO.") paths.append( { "id": PATH_DEBT, "label": "Dłużne / gwarancje (BGK, ARP)", "score": debt_score, "reasons": debt_reasons, "primary": False, "de_minimis_blocked": bool(dm.get("exhausted")), } ) # Direct EU prep eu_score = 40.0 eu_reasons = ["Direct EU (HE/EIC) lub Granty na Eurogranty jako przygotowanie."] if any(k in goal_l for k in ("horizon", "eic", "deeptech", "cleantech", "eurogrant", "horyzont")): eu_score += 30 eu_reasons.append("Cel deeptech/UE — rozważ Horyzont/EIC i Eurogranty prep.") paths.append( { "id": PATH_DIRECT_EU, "label": "Direct EU / przygotowanie Eurograntu", "score": eu_score, "reasons": eu_reasons, "primary": False, } ) # Sort and mark primary (highest that is not de_minimis_blocked when exhausted) paths.sort(key=lambda p: float(p.get("score") or 0), reverse=True) primary_set = False for p in paths: if p.get("de_minimis_blocked"): continue if not primary_set: p["primary"] = True primary_set = True break if not primary_set and paths: # all blocked weird case — still mark tax for p in paths: if p["id"] == PATH_TAX: p["primary"] = True break # Ensure ≥4 distinct path ids always present ids = {p["id"] for p in paths} assert len(ids) >= 4 # Attach tax + Direct EU checklists (F3 / F6 product surface) paths = [attach_tax_checklists(p) for p in paths] paths = [attach_direct_eu_meta(p) for p in paths] # Filter primary recommendation list: do not list de-minimis-blocked as recommended primary recommended = [p for p in paths if not p.get("de_minimis_blocked")] if not recommended: recommended = [p for p in paths if p["id"] == PATH_TAX] or paths[:1] cascade_on = strategy_cascade_enabled() tax_on = tax_paths_enabled() notes = [ "Kolejność kaskady 2026: podatki → region/FST → FENG → dłużne → Direct EU.", "Wynik jest rekomendacją produktową, nie opinią prawną.", ] if not cascade_on: notes.append( "ENABLE_STRATEGY_CASCADE=false — kaskada nie jest egzekwowana (paths nadal zwracane)." ) if not tax_on: notes.append( "ENABLE_TAX_PATHS=false — checklisty PSI/ulga B+R/IP Box nie dołączone do PATH_TAX." ) return { "version": 1, "paths": paths, "recommended_paths": recommended[:4], "primary_path_id": next((p["id"] for p in paths if p.get("primary")), recommended[0]["id"]), "cascade_order": [ PATH_TAX, PATH_REGIONAL_FST, PATH_NATIONAL_FENG, PATH_DEBT, PATH_DIRECT_EU, ], "cascade_enforced": cascade_on, "tax_paths_enabled": tax_on, "eligibility_summary": { "msp": msp.get("status"), "de_minimis_exhausted": bool(dm.get("exhausted")), "de_minimis_used_eur": dm.get("used_eur"), }, "notes": notes, }