feat(ai): composite risk scoring engine with explainable factors

- ai/indicators.py: five indicator functions with Neo4j queries
politician_company_overlap (weight 0.35)
contract_concentration (weight 0.25)
audit_mention_frequency (weight 0.20)
asset_growth_anomaly (weight 0.15)
criminal_case_presence (weight 0.05)
Each returns raw_score, weighted, evidence list, source citations
- ai/explainer.py: score_to_level(), generate_explanation(),
validate_language() enforcing FORBIDDEN_WORDS list
Never outputs: corrupt, bribe, guilty, fraud, suspect
- ai/risk_scorer.py: RiskScorer class, score_entity() per entity,
score_all_politicians(), score_all_companies(), write_scores_to_graph()
CLI: --entity-id ID | --all | --all --write-graph

ai/explainer.py

@@ -0,0 +1,79 @@
+import os
+import sys
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+
+RISK_LEVELS = {
+    (0,  30):  "LOW",
+    (31, 60):  "MODERATE",
+    (61, 80):  "HIGH",
+    (81, 100): "VERY_HIGH",
+}
+
+FORBIDDEN_WORDS = [
+    "corrupt", "bribe", "guilty", "criminal", "illegal",
+    "fraud", "suspect", "launder", "steal", "theft",
+]
+
+
+def score_to_level(score: int) -> str:
+    for (lo, hi), level in RISK_LEVELS.items():
+        if lo <= score <= hi:
+            return level
+    return "UNKNOWN"
+
+
+def generate_explanation(entity_name: str, score: int,
+                          factors: list) -> str:
+    level  = score_to_level(score)
+    active = [f for f in factors if f["raw_score"] > 0]
+
+    if score == 0 or not active:
+        return (
+            f"No structural indicators were identified for {entity_name} "
+            "in the current dataset. Limited data coverage may affect this result."
+        )
+
+    factor_names = [f["name"].replace("_", " ") for f in active]
+    factor_list  = ", ".join(factor_names)
+
+    if level == "LOW":
+        return (
+            f"Low structural indicators detected for {entity_name}. "
+            f"{len(active)} pattern(s) identified: {factor_list}. "
+            "Signals are within normal range for publicly available data."
+        )
+    if level == "MODERATE":
+        return (
+            f"Moderate structural indicators detected for {entity_name}. "
+            f"{len(active)} pattern(s) identified: {factor_list}. "
+            "Further review of the associated evidence is warranted. "
+            "This is an analytical observation based on public records."
+        )
+    if level == "HIGH":
+        return (
+            f"High structural indicators detected for {entity_name}. "
+            f"{len(active)} significant pattern(s) identified: {factor_list}. "
+            "The combination of these indicators across procurement and audit "
+            "data warrants investigative attention. "
+            "This is an analytical indicator derived from official public records, "
+            "not a legal finding."
+        )
+    return (
+        f"Very high structural indicators detected for {entity_name}. "
+        f"{len(active)} strong pattern(s) identified: {factor_list}. "
+        "Multiple overlapping patterns across procurement, audit, and disclosure "
+        "data are present. All findings are based on official public records. "
+        "This is a statistical structural indicator, not a legal determination."
+    )
+
+
+def validate_language(text: str) -> str:
+    text_lower = text.lower()
+    for word in FORBIDDEN_WORDS:
+        if word in text_lower:
+            raise ValueError(
+                f"Output contains prohibited accusatory term '{word}'. "
+                "Revise to use neutral analytical language."
+            )
+    return text

ai/indicators.py

@@ -0,0 +1,188 @@
+import os
+import sys
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from loguru import logger
+
+
+WEIGHTS = {
+    "politician_company_overlap": 0.35,
+    "contract_concentration":     0.25,
+    "audit_mention_frequency":    0.20,
+    "asset_growth_anomaly":       0.15,
+    "criminal_case_presence":     0.05,
+}
+
+MAX_SCORE = 100
+
+
+def indicator_politician_company_overlap(entity_id: str, session) -> dict:
+    row = session.run(
+        """
+        MATCH (p {id: $id})-[:DIRECTOR_OF]->(c:Company)-[:WON_CONTRACT]->(ct:Contract)
+        RETURN count(ct) AS cnt, sum(ct.amount_crore) AS total
+        """,
+        id=entity_id
+    ).single()
+
+    cnt   = row["cnt"]   if row and row["cnt"]   else 0
+    total = row["total"] if row and row["total"] else 0.0
+
+    raw   = min(int(cnt * 10), 35)
+    return {
+        "name":        "politician_company_overlap",
+        "raw_score":   raw,
+        "weight":      WEIGHTS["politician_company_overlap"],
+        "weighted":    round(raw * WEIGHTS["politician_company_overlap"], 2),
+        "description": (
+            f"Entity linked to {cnt} contract(s) totalling Rs {total:.1f} Cr "
+            "through company directorships."
+        ),
+        "evidence": [
+            f"{cnt} contract(s) found via DIRECTOR_OF -> WON_CONTRACT path",
+            f"Total contract value: Rs {round(total, 2)} Cr",
+            "Source: Government e-Marketplace procurement records",
+        ],
+        "source_institution": "Government e-Marketplace",
+        "source_url":         "https://gem.gov.in",
+    }
+
+
+def indicator_contract_concentration(entity_id: str, session) -> dict:
+    row = session.run(
+        """
+        MATCH (c {id: $id})-[:WON_CONTRACT]->(ct:Contract)
+        RETURN count(ct) AS cnt, sum(ct.amount_crore) AS total
+        """,
+        id=entity_id
+    ).single()
+
+    cnt   = row["cnt"]   if row and row["cnt"]   else 0
+    total = row["total"] if row and row["total"] else 0.0
+
+    raw   = min(int(cnt * 8), 25)
+    return {
+        "name":        "contract_concentration",
+        "raw_score":   raw,
+        "weight":      WEIGHTS["contract_concentration"],
+        "weighted":    round(raw * WEIGHTS["contract_concentration"], 2),
+        "description": (
+            f"Entity awarded {cnt} government contract(s) totalling Rs {total:.1f} Cr. "
+            "Repeated awards to the same entity indicate concentration."
+        ),
+        "evidence": [
+            f"{cnt} contract(s) via WON_CONTRACT relationships",
+            f"Total value: Rs {round(total, 2)} Cr",
+            "Source: Government e-Marketplace procurement records",
+        ],
+        "source_institution": "Government e-Marketplace",
+        "source_url":         "https://gem.gov.in",
+    }
+
+
+def indicator_audit_mention_frequency(entity_id: str, entity_name: str,
+                                       session) -> dict:
+    row = session.run(
+        """
+        MATCH (a:AuditReport)
+        WHERE toLower(a.title) CONTAINS toLower($name)
+        RETURN count(a) AS cnt, sum(a.amount_crore) AS total
+        """,
+        name=entity_name
+    ).single()
+
+    cnt   = row["cnt"]   if row and row["cnt"]   else 0
+    total = row["total"] if row and row["total"] else 0.0
+
+    raw   = min(int(cnt * 10), 20)
+    return {
+        "name":        "audit_mention_frequency",
+        "raw_score":   raw,
+        "weight":      WEIGHTS["audit_mention_frequency"],
+        "weighted":    round(raw * WEIGHTS["audit_mention_frequency"], 2),
+        "description": (
+            f"Entity or associated names appear in {cnt} CAG audit report(s). "
+            f"Total amount flagged in those reports: Rs {total:.1f} Cr."
+        ),
+        "evidence": [
+            f"{cnt} CAG report mention(s)",
+            f"Total flagged amount: Rs {round(total, 2)} Cr",
+            "Source: Comptroller and Auditor General of India, cag.gov.in",
+        ],
+        "source_institution": "Comptroller and Auditor General of India",
+        "source_url":         "https://cag.gov.in/en/audit-report",
+    }
+
+
+def indicator_asset_growth_anomaly(entity_id: str, session) -> dict:
+    row = session.run(
+        """
+        MATCH (p:Politician {id: $id})
+        RETURN p.total_assets AS assets
+        """,
+        id=entity_id
+    ).single()
+
+    assets_str = row["assets"] if row else ""
+    raw = 0
+    description = "Insufficient asset declaration data for growth analysis."
+
+    if assets_str and any(c.isdigit() for c in str(assets_str)):
+        raw = 5
+        description = (
+            "Asset declaration data available from election affidavit. "
+            "Multi-cycle comparison requires affidavit data from consecutive elections."
+        )
+
+    return {
+        "name":        "asset_growth_anomaly",
+        "raw_score":   raw,
+        "weight":      WEIGHTS["asset_growth_anomaly"],
+        "weighted":    round(raw * WEIGHTS["asset_growth_anomaly"], 2),
+        "description": description,
+        "evidence": [
+            f"Declared assets: {assets_str or 'not available'}",
+            "Source: Election Commission of India candidate affidavit",
+        ],
+        "source_institution": "Election Commission of India",
+        "source_url":         "https://myneta.info",
+    }
+
+
+def indicator_criminal_case_presence(entity_id: str, session) -> dict:
+    row = session.run(
+        """
+        MATCH (p:Politician {id: $id})
+        RETURN toInteger(p.criminal_cases) AS cases
+        """,
+        id=entity_id
+    ).single()
+
+    cases = row["cases"] if row and row["cases"] else 0
+    raw   = min(int(cases * 3), 5)
+
+    return {
+        "name":        "criminal_case_presence",
+        "raw_score":   raw,
+        "weight":      WEIGHTS["criminal_case_presence"],
+        "weighted":    round(raw * WEIGHTS["criminal_case_presence"], 2),
+        "description": (
+            f"Entity has declared {cases} criminal case(s) in their "
+            "Election Commission of India candidate affidavit."
+        ),
+        "evidence": [
+            f"{cases} declared criminal case(s)",
+            "Source: Election Commission of India candidate affidavit (self-declared)",
+        ],
+        "source_institution": "Election Commission of India",
+        "source_url":         "https://eci.gov.in",
+    }
+
+
+ALL_INDICATORS = [
+    indicator_politician_company_overlap,
+    indicator_contract_concentration,
+    indicator_audit_mention_frequency,
+    indicator_asset_growth_anomaly,
+    indicator_criminal_case_presence,
+]

ai/risk_scorer.py

@@ -0,0 +1,218 @@
+import os
+import sys
+import json
+import argparse
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from datetime import datetime
+from loguru import logger
+
+from ai.indicators import (
+    indicator_politician_company_overlap,
+    indicator_contract_concentration,
+    indicator_audit_mention_frequency,
+    indicator_asset_growth_anomaly,
+    indicator_criminal_case_presence,
+)
+from ai.explainer import score_to_level, generate_explanation, validate_language
+
+
+class RiskScorer:
+
+    def __init__(self):
+        self.driver = None
+        self._connect()
+
+    def _connect(self):
+        try:
+            from neo4j import GraphDatabase
+            from dotenv import load_dotenv
+            load_dotenv()
+            uri  = os.getenv("NEO4J_URI", "")
+            user = os.getenv("NEO4J_USER", "neo4j")
+            pwd  = os.getenv("NEO4J_PASSWORD", "")
+            if not uri or not pwd:
+                raise ValueError("NEO4J_URI and NEO4J_PASSWORD must be set in .env")
+            self.driver = GraphDatabase.driver(uri, auth=(user, pwd))
+            self.driver.verify_connectivity()
+            logger.success("[RiskScorer] Connected to Neo4j")
+        except Exception as e:
+            logger.error(f"[RiskScorer] Connection failed: {e}")
+            raise
+
+    def score_entity(self, entity_id: str) -> dict:
+        with self.driver.session() as session:
+            entity = session.run(
+                "MATCH (n {id: $id}) RETURN n.name AS name, labels(n)[0] AS type",
+                id=entity_id
+            ).single()
+
+            if not entity:
+                return {
+                    "entity_id":    entity_id,
+                    "entity_name":  entity_id,
+                    "entity_type":  "unknown",
+                    "risk_score":   0,
+                    "risk_level":   "UNKNOWN",
+                    "factors":      [],
+                    "explanation":  "Entity not found in graph.",
+                    "scored_at":    datetime.now().isoformat(),
+                }
+
+            entity_name = entity["name"] or entity_id
+            entity_type = entity["type"] or "unknown"
+
+            factors = [
+                indicator_politician_company_overlap(entity_id, session),
+                indicator_contract_concentration(entity_id, session),
+                indicator_audit_mention_frequency(entity_id, entity_name, session),
+                indicator_asset_growth_anomaly(entity_id, session),
+                indicator_criminal_case_presence(entity_id, session),
+            ]
+
+            total = sum(f["raw_score"] for f in factors)
+            score = min(total, 100)
+            level = score_to_level(score)
+            explanation = validate_language(
+                generate_explanation(entity_name, score, factors)
+            )
+
+            logger.info(
+                f"[RiskScorer] {entity_name}: score={score} level={level} "
+                f"factors={len([f for f in factors if f['raw_score'] > 0])}"
+            )
+
+            return {
+                "entity_id":   entity_id,
+                "entity_name": entity_name,
+                "entity_type": entity_type,
+                "risk_score":  score,
+                "risk_level":  level,
+                "factors":     factors,
+                "explanation": explanation,
+                "sources": list({
+                    f["source_institution"]: f["source_url"]
+                    for f in factors
+                }.items()),
+                "scored_at": datetime.now().isoformat(),
+            }
+
+    def score_all_politicians(self, limit: int = 50) -> list:
+        with self.driver.session() as session:
+            rows = session.run(
+                "MATCH (p:Politician) RETURN p.id AS id LIMIT $limit",
+                limit=limit
+            ).data()
+
+        results = []
+        for row in rows:
+            try:
+                result = self.score_entity(row["id"])
+                results.append(result)
+            except Exception as e:
+                logger.error(f"[RiskScorer] Failed to score {row['id']}: {e}")
+
+        results.sort(key=lambda x: x["risk_score"], reverse=True)
+        logger.success(f"[RiskScorer] Scored {len(results)} politicians")
+        return results
+
+    def score_all_companies(self, limit: int = 50) -> list:
+        with self.driver.session() as session:
+            rows = session.run(
+                "MATCH (c:Company) RETURN c.id AS id LIMIT $limit",
+                limit=limit
+            ).data()
+
+        results = []
+        for row in rows:
+            try:
+                result = self.score_entity(row["id"])
+                results.append(result)
+            except Exception as e:
+                logger.error(f"[RiskScorer] Failed to score {row['id']}: {e}")
+
+        results.sort(key=lambda x: x["risk_score"], reverse=True)
+        logger.success(f"[RiskScorer] Scored {len(results)} companies")
+        return results
+
+    def save_scores(self, scores: list, filepath: str):
+        os.makedirs(os.path.dirname(filepath), exist_ok=True)
+        with open(filepath, "w", encoding="utf-8") as f:
+            json.dump(scores, f, indent=2, ensure_ascii=False)
+        logger.success(f"[RiskScorer] Saved {len(scores)} scores to {filepath}")
+
+    def write_scores_to_graph(self, scores: list):
+        with self.driver.session() as session:
+            for s in scores:
+                session.run(
+                    """
+                    MATCH (n {id: $id})
+                    SET n.risk_score = $score,
+                        n.risk_level = $level,
+                        n.scored_at  = $at
+                    """,
+                    id=s["entity_id"],
+                    score=s["risk_score"],
+                    level=s["risk_level"],
+                    at=s["scored_at"],
+                )
+        logger.success(f"[RiskScorer] Wrote {len(scores)} scores to Neo4j nodes")
+
+    def close(self):
+        if self.driver:
+            self.driver.close()
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="BharatGraph Risk Scorer")
+    parser.add_argument("--entity-id", type=str, default=None)
+    parser.add_argument("--all", action="store_true")
+    parser.add_argument("--write-graph", action="store_true")
+    args = parser.parse_args()
+
+    print("=" * 55)
+    print("BharatGraph - Risk Scoring Engine")
+    print("=" * 55)
+
+    scorer = RiskScorer()
+
+    if args.entity_id:
+        result = scorer.score_entity(args.entity_id)
+        print(f"\nEntity:      {result['entity_name']}")
+        print(f"Type:        {result['entity_type']}")
+        print(f"Risk Score:  {result['risk_score']} / 100")
+        print(f"Risk Level:  {result['risk_level']}")
+        print(f"\nExplanation:\n  {result['explanation']}")
+        print(f"\nFactors:")
+        for f in result["factors"]:
+            if f["raw_score"] > 0:
+                print(f"  {f['name']}: {f['raw_score']} pts")
+                print(f"    {f['description']}")
+
+    elif args.all:
+        politicians = scorer.score_all_politicians(limit=100)
+        companies   = scorer.score_all_companies(limit=100)
+        all_scores  = politicians + companies
+
+        if args.write_graph:
+            scorer.write_scores_to_graph(all_scores)
+
+        from datetime import datetime as dt
+        ts = dt.now().strftime("%Y%m%d_%H%M%S")
+        scorer.save_scores(all_scores, f"data/processed/risk_scores_{ts}.json")
+
+        print(f"\nTotal scored: {len(all_scores)}")
+        print(f"High or above: {len([s for s in all_scores if s['risk_score'] >= 61])}")
+        print(f"With any indicator: {len([s for s in all_scores if s['risk_score'] > 0])}")
+
+        if all_scores:
+            print(f"\nTop results:")
+            for s in all_scores[:5]:
+                print(f"  {s['entity_name']:30s} {s['risk_score']:3d}  {s['risk_level']}")
+    else:
+        print("\nUsage:")
+        print("  python -m ai.risk_scorer --entity-id <id>")
+        print("  python -m ai.risk_scorer --all")
+        print("  python -m ai.risk_scorer --all --write-graph")
+
+    scorer.close()