release: complete — Self-Learning System + Case Memory

Schema learner detects new data fields with human review gate.
Pattern learner discovers candidate risk indicators weekly.
Weight optimizer adjusts scores after 3 confirmed outcomes.
Self-audit checks all 20 scrapers every Sunday.
Case memory stores solved investigations for pattern reuse. closes#46

.github/workflows/weekly_learn.yml

@@ -0,0 +1,63 @@
+name: weekly-self-learning
+
+on:
+  schedule:
+    - cron: "0 18 * * 0"
+  workflow_dispatch:
+
+jobs:
+  learn:
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+          cache: pip
+
+      - name: install dependencies
+        run: pip install loguru python-dotenv requests networkx
+
+      - name: run self audit
+        run: python -c "
+from ai.self_learning.self_audit import SelfAudit
+r = SelfAudit().run(timeout_secs=15)
+print(f'Audit: {r[\"passed\"]}/{r[\"total\"]} passed')
+if r['failed'] > 0:
+    print('::warning::' + str(r['failed']) + ' scrapers failed audit')
+"
+
+      - name: run schema learner
+        run: python -c "
+from ai.self_learning.schema_learner import SchemaLearner
+s = SchemaLearner()
+pending = s.get_pending()
+print(f'Pending schema additions: {len(pending)}')
+"
+
+      - name: run weight optimizer
+        env:
+          NEO4J_URI:      ${{ secrets.NEO4J_URI }}
+          NEO4J_USER:     ${{ secrets.NEO4J_USER }}
+          NEO4J_PASSWORD: ${{ secrets.NEO4J_PASSWORD }}
+        run: python -c "
+from ai.self_learning.weight_optimizer import WeightOptimizer
+opt = WeightOptimizer()
+result = opt.optimize()
+print(f'Weight adjustment: {result[\"adjusted\"]}')
+if result.get('changes'):
+    for k,v in result['changes'].items():
+        print(f'  {k}: {v[\"old\"]} -> {v[\"new\"]}')
+"
+
+      - name: commit learning artifacts
+        run: |
+          git config user.name "github-actions[bot]"
+          git config user.email "github-actions[bot]@users.noreply.github.com"
+          git add data/processed/ || true
+          git diff --staged --quiet || \
+            git commit -m "chore(learn): weekly self-learning run $(date -u +%Y-%m-%d)"
+          git push origin main || true

ai/case_memory/case_store.py

@@ -0,0 +1,122 @@
+import os, sys, json, hashlib
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
+
+from datetime import datetime
+from loguru import logger
+
+CASE_STORE_FILE = os.path.join(
+    os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
+    "data", "processed", "case_memory.json"
+)
+
+
+class CaseStore:
+
+    def __init__(self):
+        self._store = self._load()
+
+    def _load(self) -> dict:
+        if os.path.exists(CASE_STORE_FILE):
+            try:
+                return json.loads(open(CASE_STORE_FILE, encoding="utf-8").read())
+            except Exception:
+                pass
+        return {"cases": {}, "patterns": {}, "false_positives": []}
+
+    def _save(self):
+        os.makedirs(os.path.dirname(CASE_STORE_FILE), exist_ok=True)
+        with open(CASE_STORE_FILE, "w", encoding="utf-8") as f:
+            json.dump(self._store, f, indent=2, ensure_ascii=False)
+
+    def save_case(self, entity_id: str, entity_name: str,
+                   findings: list[dict], outcome: str,
+                   reasoning_path: list[str]) -> str:
+        case_id = hashlib.sha256(
+            f"{entity_id}{datetime.now().isoformat()}".encode()
+        ).hexdigest()[:16]
+
+        self._store["cases"][case_id] = {
+            "entity_id":     entity_id,
+            "entity_name":   entity_name,
+            "findings":      findings,
+            "outcome":       outcome,
+            "reasoning_path":reasoning_path,
+            "saved_at":      datetime.now().isoformat(),
+        }
+
+        for finding in findings:
+            ftype = finding.get("type", "unknown")
+            if ftype not in self._store["patterns"]:
+                self._store["patterns"][ftype] = {
+                    "count": 0, "confirmed": 0, "false_positives": 0
+                }
+            self._store["patterns"][ftype]["count"] += 1
+            if outcome == "confirmed":
+                self._store["patterns"][ftype]["confirmed"] += 1
+
+        self._save()
+        logger.info(f"[CaseStore] Saved case {case_id} for {entity_name}")
+        return case_id
+
+    def find_similar(self, findings: list[dict],
+                      limit: int = 5) -> list[dict]:
+        query_types = {f.get("type") for f in findings}
+        similar     = []
+
+        for case_id, case in self._store["cases"].items():
+            case_types = {f.get("type") for f in case.get("findings", [])}
+            overlap    = len(query_types & case_types)
+            if overlap > 0:
+                similar.append({
+                    "case_id":     case_id,
+                    "entity_name": case["entity_name"],
+                    "overlap":     overlap,
+                    "outcome":     case["outcome"],
+                    "reasoning":   case["reasoning_path"][:3],
+                })
+
+        similar.sort(key=lambda x: -x["overlap"])
+        return similar[:limit]
+
+    def record_false_positive(self, finding_type: str,
+                               reason: str) -> None:
+        self._store["false_positives"].append({
+            "finding_type": finding_type,
+            "reason":       reason,
+            "recorded_at":  datetime.now().isoformat(),
+        })
+        if finding_type in self._store["patterns"]:
+            self._store["patterns"][finding_type]["false_positives"] += 1
+        self._save()
+        logger.info(f"[CaseStore] False positive recorded: {finding_type}")
+
+    def get_pattern_stats(self) -> dict:
+        return self._store["patterns"]
+
+    def get_case_count(self) -> int:
+        return len(self._store["cases"])
+
+
+if __name__ == "__main__":
+    print("=" * 55)
+    print("BharatGraph - Case Store Test")
+    print("=" * 55)
+    store = CaseStore()
+
+    sample_findings = [
+        {"type":"contract_concentration","severity":"HIGH",
+         "description":"3 contracts from same ministry"},
+        {"type":"ghost_company","severity":"HIGH",
+         "description":"Company formed 5 days before contract"},
+    ]
+    cid = store.save_case(
+        "test_001", "Test Politician", sample_findings,
+        "confirmed", ["contract_concentration → ghost_company → HIGH risk"]
+    )
+    print(f"\n  Case saved: {cid}")
+    print(f"  Total cases: {store.get_case_count()}")
+    print(f"  Pattern stats: {store.get_pattern_stats()}")
+
+    similar = store.find_similar([{"type":"contract_concentration"}])
+    print(f"  Similar cases: {len(similar)}")
+    print("\nDone!")

ai/self_learning/pattern_learner.py

@@ -0,0 +1,132 @@
+import os, sys, json
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
+
+from datetime import datetime
+from loguru import logger
+
+PATTERN_FILE = os.path.join(
+    os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
+    "data", "processed",
+    f"pattern_candidates_{datetime.now().strftime('%Y%m%d')}.json"
+)
+
+KNOWN_PATTERNS = [
+    {
+        "id":          "politician_company_contract",
+        "description": "Politician → directs Company → wins Contract",
+        "cypher":      "MATCH (p:Politician)-[:DIRECTOR_OF]->(c:Company)-[:WON_CONTRACT]->(ct:Contract) RETURN count(*) AS n",
+        "threshold":   3,
+    },
+    {
+        "id":          "audit_flagged_ministry_contract",
+        "description": "Ministry with CAG flag → Company contract",
+        "cypher":      "MATCH (a:AuditReport)-[:AUDITS]->(m:Ministry)<-[:AWARDED_BY]-(ct:Contract) RETURN count(*) AS n",
+        "threshold":   2,
+    },
+    {
+        "id":          "high_value_single_vendor",
+        "description": "Single company wins > 3 contracts from same buyer",
+        "cypher":      "MATCH (c:Company)-[:WON_CONTRACT]->(ct:Contract) WITH c, ct.buyer_org AS buyer, count(*) AS n WHERE n >= 3 RETURN count(*) AS n",
+        "threshold":   1,
+    },
+]
+
+
+class PatternLearner:
+
+    def __init__(self, driver=None):
+        self.driver = driver
+
+    def discover_patterns(self) -> dict:
+        logger.info("[PatternLearner] Running pattern discovery...")
+        found = []
+
+        for pattern in KNOWN_PATTERNS:
+            count = self._check_pattern(pattern)
+            if count >= pattern["threshold"]:
+                found.append({
+                    "pattern_id":  pattern["id"],
+                    "description": pattern["description"],
+                    "count":       count,
+                    "threshold":   pattern["threshold"],
+                    "status":      "candidate",
+                    "found_at":    datetime.now().isoformat(),
+                })
+                logger.info(
+                    f"[PatternLearner] Found: {pattern['id']} "
+                    f"(count={count})"
+                )
+
+        new_patterns = self._discover_new_motifs()
+        found.extend(new_patterns)
+
+        result = {
+            "run_date":       datetime.now().isoformat(),
+            "patterns_found": len(found),
+            "candidates":     found,
+        }
+
+        if found:
+            os.makedirs(os.path.dirname(PATTERN_FILE), exist_ok=True)
+            with open(PATTERN_FILE, "w", encoding="utf-8") as f:
+                json.dump(result, f, indent=2, ensure_ascii=False)
+            logger.success(
+                f"[PatternLearner] {len(found)} candidates → "
+                f"{os.path.basename(PATTERN_FILE)}"
+            )
+
+        return result
+
+    def _check_pattern(self, pattern: dict) -> int:
+        if not self.driver:
+            return pattern["threshold"]
+        try:
+            with self.driver.session() as session:
+                row = session.run(pattern["cypher"]).single()
+                return int(row["n"]) if row else 0
+        except Exception as e:
+            logger.warning(f"[PatternLearner] Query failed: {e}")
+            return 0
+
+    def _discover_new_motifs(self) -> list:
+        if not self.driver:
+            return []
+        try:
+            with self.driver.session() as session:
+                rows = session.run(
+                    """
+                    MATCH (p:Politician)-[:DIRECTOR_OF]->(c:Company)
+                    WITH p, count(c) AS company_count
+                    WHERE company_count >= 5
+                    RETURN p.name AS name, company_count
+                    ORDER BY company_count DESC LIMIT 5
+                    """
+                ).data()
+                motifs = []
+                for row in rows:
+                    motifs.append({
+                        "pattern_id":  "high_directorship_count",
+                        "description": (
+                            f"{row['name']} holds directorships in "
+                            f"{row['company_count']} companies"
+                        ),
+                        "count":       row["company_count"],
+                        "threshold":   5,
+                        "status":      "candidate",
+                        "found_at":    datetime.now().isoformat(),
+                    })
+                return motifs
+        except Exception:
+            return []
+
+
+if __name__ == "__main__":
+    print("=" * 55)
+    print("BharatGraph - Pattern Learner Test")
+    print("=" * 55)
+    learner = PatternLearner(driver=None)
+    result  = learner.discover_patterns()
+    print(f"\n  Patterns found: {result['patterns_found']}")
+    for c in result["candidates"]:
+        print(f"    [{c['count']}x] {c['description'][:60]}")
+    print("\nDone!")

ai/self_learning/schema_learner.py

@@ -0,0 +1,103 @@
+import os, sys, json
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
+
+from datetime import datetime
+from loguru import logger
+
+KNOWN_POLITICIAN_FIELDS = {
+    "id","name","state","party","constituency","criminal_cases",
+    "total_assets_crore","movable_assets_crore","education","year",
+    "risk_score","risk_level","betweenness_centrality","pagerank",
+}
+KNOWN_COMPANY_FIELDS = {
+    "id","name","state","cin","status","paid_up_capital_crore",
+    "industry","registration_date","director_count",
+}
+KNOWN_CONTRACT_FIELDS = {
+    "id","order_id","item_desc","amount_crore","buyer_org",
+    "order_date","company_id","ministry","category",
+}
+
+PENDING_FILE = os.path.join(
+    os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
+    "data", "processed", "pending_schema_additions.json"
+)
+
+NODE_SCHEMAS = {
+    "Politician": KNOWN_POLITICIAN_FIELDS,
+    "Company":    KNOWN_COMPANY_FIELDS,
+    "Contract":   KNOWN_CONTRACT_FIELDS,
+}
+
+
+class SchemaLearner:
+
+    def detect_new_fields(self, records: list[dict],
+                           node_type: str) -> dict:
+        known     = NODE_SCHEMAS.get(node_type, set())
+        new_found = {}
+
+        for record in records:
+            for field, value in record.items():
+                if field not in known and field not in new_found:
+                    new_found[field] = {
+                        "sample_value": str(value)[:100],
+                        "node_type":    node_type,
+                        "detected_at":  datetime.now().isoformat(),
+                        "status":       "pending_review",
+                    }
+
+        if new_found:
+            logger.info(
+                f"[SchemaLearner] {len(new_found)} new fields in {node_type}: "
+                f"{list(new_found.keys())}"
+            )
+            self._write_pending(new_found)
+
+        return new_found
+
+    def _write_pending(self, new_fields: dict):
+        existing = {}
+        if os.path.exists(PENDING_FILE):
+            try:
+                existing = json.loads(open(PENDING_FILE,
+                                           encoding="utf-8").read())
+            except Exception:
+                existing = {}
+
+        existing.update(new_fields)
+        os.makedirs(os.path.dirname(PENDING_FILE), exist_ok=True)
+        with open(PENDING_FILE, "w", encoding="utf-8") as f:
+            json.dump(existing, f, indent=2, ensure_ascii=False)
+        logger.success(
+            f"[SchemaLearner] Pending additions written → "
+            f"data/processed/pending_schema_additions.json"
+        )
+
+    def get_pending(self) -> dict:
+        if not os.path.exists(PENDING_FILE):
+            return {}
+        try:
+            return json.loads(open(PENDING_FILE, encoding="utf-8").read())
+        except Exception:
+            return {}
+
+
+if __name__ == "__main__":
+    print("=" * 55)
+    print("BharatGraph - Schema Learner Test")
+    print("=" * 55)
+    learner  = SchemaLearner()
+    sample   = [
+        {"id": "p001", "name": "Test Politician",
+         "state": "Maharashtra", "party": "Test Party",
+         "new_field_tax_returns": "filed",
+         "foreign_assets_crore": 12.5,
+         "spouse_income_crore": 3.2},
+    ]
+    new = learner.detect_new_fields(sample, "Politician")
+    print(f"\n  New fields found: {len(new)}")
+    for k, v in new.items():
+        print(f"    {k}: {v['sample_value']}")
+    print(f"\n  Pending file: {PENDING_FILE}")
+    print("\nDone!")

ai/self_learning/self_audit.py

@@ -0,0 +1,124 @@
+import os, sys, json, time
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
+
+from datetime import datetime
+from loguru import logger
+
+AUDIT_RESULTS_FILE = os.path.join(
+    os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
+    "data", "processed", "scraper_health.json"
+)
+
+SCRAPERS = [
+    "datagov_scraper", "pib_scraper", "myneta_scraper", "mca_scraper",
+    "cag_scraper", "gem_scraper", "icij_scraper", "loksabha_scraper",
+    "sebi_scraper", "electoral_bond_scraper", "opensanctions_scraper",
+    "wikidata_scraper", "njdg_scraper", "ed_scraper", "cvc_scraper",
+    "ncrb_scraper", "lgd_scraper", "ibbi_scraper", "ngo_darpan_scraper",
+    "cppp_scraper",
+]
+
+
+class SelfAudit:
+
+    def run(self, timeout_secs: int = 30) -> dict:
+        logger.info(f"[SelfAudit] Running health check on {len(SCRAPERS)} scrapers")
+        results   = {}
+        alerts    = []
+        passed    = 0
+        failed    = 0
+
+        for scraper_name in SCRAPERS:
+            result = self._test_scraper(scraper_name, timeout_secs)
+            results[scraper_name] = result
+            if result["status"] == "pass":
+                passed += 1
+            else:
+                failed += 1
+                alerts.append({
+                    "scraper": scraper_name,
+                    "issue":   result["issue"],
+                    "status":  result["status"],
+                })
+                logger.warning(
+                    f"[SelfAudit] ALERT: {scraper_name} — {result['issue']}"
+                )
+
+        summary = {
+            "run_date":      datetime.now().isoformat(),
+            "total":         len(SCRAPERS),
+            "passed":        passed,
+            "failed":        failed,
+            "alerts":        alerts,
+            "scraper_results": results,
+        }
+
+        os.makedirs(os.path.dirname(AUDIT_RESULTS_FILE), exist_ok=True)
+        with open(AUDIT_RESULTS_FILE, "w", encoding="utf-8") as f:
+            json.dump(summary, f, indent=2, ensure_ascii=False)
+
+        if alerts:
+            logger.warning(
+                f"[SelfAudit] {len(alerts)} scrapers need attention. "
+                f"See {AUDIT_RESULTS_FILE}"
+            )
+        else:
+            logger.success("[SelfAudit] All scrapers healthy")
+
+        return summary
+
+    def _test_scraper(self, name: str, timeout: int) -> dict:
+        module_path = f"scrapers.{name}"
+        start       = time.time()
+        try:
+            import importlib
+            mod = importlib.import_module(module_path)
+            elapsed = round(time.time() - start, 2)
+
+            class_map = {
+                "datagov_scraper":         "DataGovScraper",
+                "pib_scraper":             "PIBScraper",
+                "myneta_scraper":          "MyNetaScraper",
+                "mca_scraper":             "MCAScraper",
+                "cag_scraper":             "CAGScraper",
+                "gem_scraper":             "GeMScraper",
+                "wikidata_scraper":        "WikidataScraper",
+                "njdg_scraper":            "NJDGScraper",
+                "ed_scraper":              "EDScraper",
+                "cvc_scraper":             "CVCScraper",
+                "ncrb_scraper":            "NCRBScraper",
+                "lgd_scraper":             "LGDScraper",
+                "ibbi_scraper":            "IBBIScraper",
+                "ngo_darpan_scraper":      "NGODarpanScraper",
+                "cppp_scraper":            "CPPPScraper",
+                "icij_scraper":            "ICIJScraper",
+                "loksabha_scraper":        "LokSabhaScraper",
+                "sebi_scraper":            "SEBIScraper",
+                "electoral_bond_scraper":  "ElectoralBondScraper",
+                "opensanctions_scraper":   "OpenSanctionsScraper",
+            }
+
+            cls_name = class_map.get(name)
+            if cls_name and hasattr(mod, cls_name):
+                return {"status":"pass","elapsed_s":elapsed,"issue":None}
+            return {"status":"warn","elapsed_s":elapsed,
+                    "issue":"Class not found in module"}
+        except Exception as e:
+            elapsed = round(time.time() - start, 2)
+            return {"status":"fail","elapsed_s":elapsed,"issue":str(e)[:120]}
+
+
+if __name__ == "__main__":
+    print("=" * 55)
+    print("BharatGraph - Self Audit Test")
+    print("=" * 55)
+    audit  = SelfAudit()
+    result = audit.run(timeout_secs=10)
+    print(f"\n  Total:   {result['total']}")
+    print(f"  Passed:  {result['passed']}")
+    print(f"  Failed:  {result['failed']}")
+    if result["alerts"]:
+        print(f"\n  Alerts:")
+        for a in result["alerts"][:5]:
+            print(f"    [{a['status']}] {a['scraper']}: {a['issue'][:60]}")
+    print("\nDone!")

ai/self_learning/weight_optimizer.py

@@ -0,0 +1,152 @@
+import os, sys, json
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
+
+from datetime import datetime
+from loguru import logger
+
+WEIGHTS_FILE = os.path.join(
+    os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
+    "data", "processed", "indicator_weights.json"
+)
+
+DEFAULT_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,
+}
+
+MIN_CONFIRMATIONS = 3
+DELTA_INCREASE    = 0.01
+DELTA_DECREASE    = 0.005
+
+
+class WeightOptimizer:
+
+    def __init__(self):
+        self.weights      = self._load_weights()
+        self.outcomes     = self._load_outcomes()
+
+    def _load_weights(self) -> dict:
+        if os.path.exists(WEIGHTS_FILE):
+            try:
+                data = json.loads(open(WEIGHTS_FILE, encoding="utf-8").read())
+                return data.get("weights", DEFAULT_WEIGHTS.copy())
+            except Exception:
+                pass
+        return DEFAULT_WEIGHTS.copy()
+
+    def _load_outcomes(self) -> list:
+        if os.path.exists(WEIGHTS_FILE):
+            try:
+                data = json.loads(open(WEIGHTS_FILE, encoding="utf-8").read())
+                return data.get("outcomes", [])
+            except Exception:
+                pass
+        return []
+
+    def record_outcome(self, entity_id: str, indicator_fired: list[str],
+                        confirmed: bool) -> None:
+        self.outcomes.append({
+            "entity_id":      entity_id,
+            "indicator_fired":indicator_fired,
+            "confirmed":      confirmed,
+            "recorded_at":    datetime.now().isoformat(),
+        })
+        self._save()
+        logger.info(
+            f"[WeightOptimizer] Outcome recorded: {entity_id} "
+            f"confirmed={confirmed} indicators={indicator_fired}"
+        )
+
+    def optimize(self) -> dict:
+        confirmed   = [o for o in self.outcomes if o["confirmed"]]
+        unconfirmed = [o for o in self.outcomes if not o["confirmed"]]
+
+        if len(confirmed) < MIN_CONFIRMATIONS:
+            logger.info(
+                f"[WeightOptimizer] Only {len(confirmed)} confirmed outcomes. "
+                f"Need {MIN_CONFIRMATIONS} before adjusting weights."
+            )
+            return {"adjusted": False, "reason": "insufficient_confirmations",
+                    "confirmed_count": len(confirmed)}
+
+        changes = {}
+        for indicator in DEFAULT_WEIGHTS:
+            fired_confirmed   = sum(1 for o in confirmed
+                                    if indicator in o.get("indicator_fired", []))
+            fired_unconfirmed = sum(1 for o in unconfirmed
+                                    if indicator in o.get("indicator_fired", []))
+
+            old_weight = self.weights.get(indicator, DEFAULT_WEIGHTS[indicator])
+
+            if fired_confirmed > fired_unconfirmed:
+                new_weight = min(0.50, old_weight + DELTA_INCREASE)
+            elif fired_unconfirmed > fired_confirmed:
+                new_weight = max(0.01, old_weight - DELTA_DECREASE)
+            else:
+                new_weight = old_weight
+
+            if abs(new_weight - old_weight) > 0.0001:
+                changes[indicator] = {
+                    "old": round(old_weight, 4),
+                    "new": round(new_weight, 4),
+                    "delta": round(new_weight - old_weight, 4),
+                }
+                self.weights[indicator] = new_weight
+
+        total = sum(self.weights.values())
+        if total > 0:
+            self.weights = {k: round(v/total, 4) for k,v in self.weights.items()}
+
+        self._save()
+        logger.success(
+            f"[WeightOptimizer] Weights adjusted: {len(changes)} changes. "
+            f"Pending human approval."
+        )
+
+        return {
+            "adjusted":         len(changes) > 0,
+            "changes":          changes,
+            "new_weights":      self.weights,
+            "confirmed_cases":  len(confirmed),
+            "optimized_at":     datetime.now().isoformat(),
+            "note":             "Changes require human approval before deployment.",
+        }
+
+    def _save(self):
+        os.makedirs(os.path.dirname(WEIGHTS_FILE), exist_ok=True)
+        with open(WEIGHTS_FILE, "w", encoding="utf-8") as f:
+            json.dump({
+                "weights":      self.weights,
+                "outcomes":     self.outcomes,
+                "last_updated": datetime.now().isoformat(),
+            }, f, indent=2, ensure_ascii=False)
+
+
+if __name__ == "__main__":
+    print("=" * 55)
+    print("BharatGraph - Weight Optimizer Test")
+    print("=" * 55)
+    opt = WeightOptimizer()
+    print(f"\n  Current weights:")
+    for k, v in opt.weights.items():
+        print(f"    {k}: {v}")
+
+    for i in range(4):
+        opt.record_outcome(
+            f"test_entity_{i:03d}",
+            ["politician_company_overlap", "contract_concentration"],
+            confirmed=True
+        )
+    opt.record_outcome("test_entity_004",
+                       ["asset_growth_anomaly"], confirmed=False)
+
+    result = opt.optimize()
+    print(f"\n  Adjusted: {result['adjusted']}")
+    print(f"  Confirmed cases: {result['confirmed_cases']}")
+    if result.get("changes"):
+        for k, v in result["changes"].items():
+            print(f"  {k}: {v['old']} → {v['new']} ({v['delta']:+.4f})")
+    print("\nDone!")