Multi-Modal Vision RAG for Spatial Liquidity

"""
ShadowChartMemory — Multi-Modal Vision RAG & Vector Memory Engine
Sovereign R&D Forge // BayesianPivot Production Mesh v2.0
"""
import json
import logging
import sqlite3
import numpy as np
import pandas as pd
from typing import Dict, Any, List, Optional, Tuple

logger = logging.getLogger("ShadowChartMemory")

class ShadowChartMemory:
    """
    Episodic Multi-Modal Vector Memory Engine.
    Stores, embeds, retrieves, and visually reasons over market geometry.
    """
    def __init__(self, db_path: str = "data/smc_alpha.db"):
        self.db_path = db_path
        self._memory_cache: List[Dict[str, Any]] = []
        self._load_episodic_memory()

    def _load_episodic_memory(self) -> None:
        """Loads historical signed trade outcomes into episodic memory."""
        try:
            conn = sqlite3.connect(self.db_path)
            conn.row_factory = sqlite3.Row
            rows = conn.execute("""
                SELECT timestamp, symbol, direction, pattern, ai_score, outcome, pnl,
                       volume_spike, true_smt, shadow_regime, hurst_val, liq_density
                FROM signed_ledger
                WHERE outcome IN ('WIN', 'LOSS', 'HIT_TP', 'HIT_SL') AND is_rogue = 0
                ORDER BY timestamp DESC LIMIT 200
            """).fetchall()
            self._memory_cache = [dict(r) for r in rows]
            conn.close()
            logger.info(f"Loaded {len(self._memory_cache)} episodic precedents into visual memory.")
        except Exception as e:
            logger.warning(f"Could not load episodic memory: {e}")

    def query_visual_precedents(
        self,
        symbol: str,
        direction: str,
        hurst_val: float,
        liq_density: float,
        top_k: int = 5
    ) -> Dict[str, Any]:
        """
        Retrieves top-K nearest historical chart precedents using multi-attribute cosine distance.
        """
        if not self._memory_cache:
            return {"historical_win_rate": 0.50, "precedent_count": 0, "confidence_bonus": 0.0}

        scores = []
        current_vector = np.array([
            1.0 if direction == "LONG" else -1.0,
            hurst_val,
            liq_density / 10.0
        ])

        for mem in self._memory_cache:
            mem_dir = 1.0 if mem.get("direction") in ("LONG", "BUY") else -1.0
            mem_hurst = float(mem.get("hurst_val", 0.50))
            mem_density = float(mem.get("liq_density", 5.0)) / 10.0
            mem_vector = np.array([mem_dir, mem_hurst, mem_density])

            # Cosine Similarity
            dot = np.dot(current_vector, mem_vector)
            norm = np.linalg.norm(current_vector) * np.linalg.norm(mem_vector)
            sim = dot / norm if norm > 0 else 0.0
            scores.append((sim, mem))

        # Sort by similarity
        scores.sort(key=lambda x: x[0], reverse=True)
        top_matches = scores[:top_k]

        wins = sum(1 for sim, m in top_matches if m.get("outcome") in ("WIN", "HIT_TP"))
        win_rate = wins / len(top_matches) if top_matches else 0.50

        # Conviction boost: +1.5 points if historical precedent win rate >= 70%
        conviction_bonus = 1.5 if win_rate >= 0.70 else (-1.5 if win_rate <= 0.30 else 0.0)

        return {
            "historical_win_rate": round(win_rate * 100, 1),
            "precedent_count": len(top_matches),
            "top_similarity": round(float(top_matches[0][0]), 3) if top_matches else 0.0,
            "conviction_bonus": conviction_bonus
        }