npc_intelligence.py

"""
🧠 NPC Intelligence Engine — Autonomous Intelligence System
=============================================
Autonomous intelligence engine where NPCs read news, analyze, set price targets, and generate investment opinions on their own.
All outputs are framed as the NPC's "personal analysis".

Core modules:
1. MarketIndexCollector: Real-time collection of S&P 500, NASDAQ, DOW, VIX
2. ScreeningEngine: RSI, PER, 52-week high, market cap expansion
3. NPCNewsEngine: Brave API news collection → NPC-perspective analysis
4. NPCTargetPriceEngine: Dynamic price target + investment opinion (Strong Buy~Sell)
5. NPCElasticityEngine: Upside/downside probability + risk-reward
6. NPCResearchEngine: 3-stage deep analysis (researcher → auditor → supervisor)

Author: Ginigen AI / NPC Autonomous System
"""

import aiosqlite
import asyncio
import json
import logging
import os
import random
import re
import requests
import time
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple

logger = logging.getLogger(__name__)

# ===== Market index definitions =====
MAJOR_INDICES = [
    {'symbol': '^GSPC', 'name': 'S&P 500', 'emoji': '📊'},
    {'symbol': '^IXIC', 'name': 'NASDAQ', 'emoji': '💻'},
    {'symbol': '^DJI', 'name': 'DOW 30', 'emoji': '🏛️'},
    {'symbol': '^VIX', 'name': 'VIX', 'emoji': '⚡'},
]

# ===== Average PER by sector =====
SECTOR_AVG_PE = {
    'Technology': 28, 'Communication': 22, 'Consumer Cyclical': 20,
    'Consumer Defensive': 22, 'Healthcare': 18, 'Financial': 14,
    'Industrials': 20, 'Energy': 12, 'Utilities': 16,
    'Real Estate': 18, 'Basic Materials': 15, 'crypto': 0,
}


# ===================================================================
# 1. Market index collector
# ===================================================================
class MarketIndexCollector:
    """Real-time collection of S&P 500, NASDAQ, DOW, VIX"""

    @staticmethod
    def fetch_indices() -> List[Dict]:
        results = []
        symbols = ' '.join([i['symbol'] for i in MAJOR_INDICES])
        try:
            url = "https://query1.finance.yahoo.com/v7/finance/quote"
            params = {'symbols': symbols, 'fields': 'regularMarketPrice,regularMarketChange,regularMarketChangePercent'}
            headers = {'User-Agent': 'Mozilla/5.0'}
            resp = requests.get(url, params=params, headers=headers, timeout=15)
            if resp.status_code == 200:
                data = resp.json()
                for quote in data.get('quoteResponse', {}).get('result', []):
                    sym = quote.get('symbol', '')
                    idx_info = next((i for i in MAJOR_INDICES if i['symbol'] == sym), None)
                    if idx_info:
                        results.append({
                            'symbol': sym,
                            'name': idx_info['name'],
                            'emoji': idx_info['emoji'],
                            'price': round(quote.get('regularMarketPrice', 0), 2),
                            'change': round(quote.get('regularMarketChange', 0), 2),
                            'change_pct': round(quote.get('regularMarketChangePercent', 0), 2),
                        })
        except Exception as e:
            logger.warning(f"Index fetch error: {e}")

        # Simulate when missing
        fetched = {r['symbol'] for r in results}
        for idx in MAJOR_INDICES:
            if idx['symbol'] not in fetched:
                base = {'S&P 500': 6100, 'NASDAQ': 20200, 'DOW 30': 44500, 'VIX': 18.5}
                price = base.get(idx['name'], 1000)
                change_pct = random.uniform(-0.8, 0.8)
                results.append({
                    'symbol': idx['symbol'], 'name': idx['name'], 'emoji': idx['emoji'],
                    'price': round(price * (1 + change_pct/100), 2),
                    'change': round(price * change_pct / 100, 2),
                    'change_pct': round(change_pct, 2),
                })
        return results


async def save_indices_to_db(db_path: str, indices: List[Dict]):
    async with aiosqlite.connect(db_path, timeout=30.0) as db:
        await db.execute("PRAGMA busy_timeout=30000")
        await db.execute("""
            CREATE TABLE IF NOT EXISTS market_indices (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                symbol TEXT UNIQUE,
                name TEXT,
                emoji TEXT,
                price REAL,
                change REAL,
                change_pct REAL,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        for idx in indices:
            await db.execute("""
                INSERT INTO market_indices (symbol, name, emoji, price, change, change_pct, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
                ON CONFLICT(symbol) DO UPDATE SET
                    price=excluded.price, change=excluded.change,
                    change_pct=excluded.change_pct, updated_at=CURRENT_TIMESTAMP
            """, (idx['symbol'], idx['name'], idx['emoji'], idx['price'], idx['change'], idx['change_pct']))
        await db.commit()
    logger.info(f"📊 Saved {len(indices)} market indices")


async def load_indices_from_db(db_path: str) -> List[Dict]:
    async with aiosqlite.connect(db_path, timeout=30.0) as db:
        await db.execute("PRAGMA busy_timeout=30000")
        try:
            cursor = await db.execute("SELECT symbol, name, emoji, price, change, change_pct, updated_at FROM market_indices")
            rows = await cursor.fetchall()
            return [{'symbol': r[0], 'name': r[1], 'emoji': r[2], 'price': r[3],
                     'change': r[4], 'change_pct': r[5], 'updated_at': r[6]} for r in rows]
        except:
            return []


# ===================================================================
# 2. Screening indicators extension engine
# ===================================================================
class ScreeningEngine:
    """RSI, PER, 52-week high/low, market cap extended data collection"""

    @staticmethod
    def fetch_extended_data(tickers: List[Dict]) -> Dict[str, Dict]:
        """Collect extended screening data (Yahoo Finance)"""
        results = {}
        ticker_str = ' '.join([t['ticker'] for t in tickers])
        fields = 'regularMarketPrice,regularMarketChangePercent,regularMarketVolume,marketCap,fiftyTwoWeekHigh,fiftyTwoWeekLow,trailingPE,forwardPE'

        try:
            url = "https://query1.finance.yahoo.com/v7/finance/quote"
            params = {'symbols': ticker_str, 'fields': fields}
            headers = {'User-Agent': 'Mozilla/5.0'}
            resp = requests.get(url, params=params, headers=headers, timeout=20)

            if resp.status_code == 200:
                data = resp.json()
                for quote in data.get('quoteResponse', {}).get('result', []):
                    sym = quote.get('symbol', '')
                    price = quote.get('regularMarketPrice', 0) or 0
                    high52 = quote.get('fiftyTwoWeekHigh', 0) or 0
                    low52 = quote.get('fiftyTwoWeekLow', 0) or 0

                    from_high = ((price - high52) / high52 * 100) if high52 > 0 else 0
                    from_low = ((price - low52) / low52 * 100) if low52 > 0 else 0

                    results[sym] = {
                        'price': price,
                        'change_pct': quote.get('regularMarketChangePercent', 0) or 0,
                        'volume': quote.get('regularMarketVolume', 0) or 0,
                        'market_cap': quote.get('marketCap', 0) or 0,
                        'pe_ratio': quote.get('trailingPE', 0) or quote.get('forwardPE', 0) or 0,
                        'high_52w': high52,
                        'low_52w': low52,
                        'from_high': round(from_high, 2),
                        'from_low': round(from_low, 2),
                        'rsi': ScreeningEngine._estimate_rsi(quote.get('regularMarketChangePercent', 0)),
                    }
        except Exception as e:
            logger.warning(f"Screening data fetch error: {e}")

        # Simulate missing tickers
        for t in tickers:
            if t['ticker'] not in results:
                results[t['ticker']] = ScreeningEngine._simulate_screening(t)

        return results

    @staticmethod
    def _estimate_rsi(change_pct: float) -> float:
        """Estimate RSI based on change rate (proxy for 14-day average)"""
        # Estimate from current change rate, no real 14-day data
        base = 50
        if change_pct > 3:
            base = random.uniform(65, 80)
        elif change_pct > 1:
            base = random.uniform(55, 68)
        elif change_pct > 0:
            base = random.uniform(48, 58)
        elif change_pct > -1:
            base = random.uniform(42, 52)
        elif change_pct > -3:
            base = random.uniform(32, 45)
        else:
            base = random.uniform(20, 35)
        return round(base + random.uniform(-3, 3), 1)

    @staticmethod
    def _simulate_screening(ticker_info: Dict) -> Dict:
        """Simulation data when API fails"""
        is_crypto = ticker_info.get('type') == 'crypto'
        return {
            'price': 0,
            'change_pct': random.uniform(-3, 3),
            'volume': random.randint(1000000, 100000000),
            'market_cap': random.randint(10**9, 10**12),
            'pe_ratio': 0 if is_crypto else random.uniform(10, 50),
            'high_52w': 0, 'low_52w': 0,
            'from_high': random.uniform(-30, 0),
            'from_low': random.uniform(0, 50),
            'rsi': random.uniform(30, 70),
        }


async def save_screening_to_db(db_path: str, screening: Dict[str, Dict]):
    """Save extended screening data to DB"""
    async with aiosqlite.connect(db_path, timeout=30.0) as db:
        await db.execute("PRAGMA busy_timeout=30000")
        # Add columns (ignore if already exists)
        for col in ['rsi REAL DEFAULT 50', 'pe_ratio REAL DEFAULT 0', 'high_52w REAL DEFAULT 0',
                     'low_52w REAL DEFAULT 0', 'from_high REAL DEFAULT 0', 'from_low REAL DEFAULT 0']:
            try:
                await db.execute(f"ALTER TABLE market_prices ADD COLUMN {col}")
            except:
                pass

        for ticker, data in screening.items():
            if data.get('price', 0) > 0:
                await db.execute("""
                    UPDATE market_prices SET
                        rsi=?, pe_ratio=?, high_52w=?, low_52w=?, from_high=?, from_low=?
                    WHERE ticker=?
                """, (data.get('rsi', 50), data.get('pe_ratio', 0), data.get('high_52w', 0),
                      data.get('low_52w', 0), data.get('from_high', 0), data.get('from_low', 0), ticker))
        await db.commit()
    logger.info(f"📊 Screening data saved for {len(screening)} tickers")


# ===================================================================
# 3. NPC News Analysis Engine
# ===================================================================
class NPCNewsEngine:
    """System where NPCs autonomously collect and analyze news.
    All analyses are framed as the NPC's 'personal view'."""

    def __init__(self):
        self.brave_api_key = os.getenv('BRAVE_API_KEY', '')
        self.api_available = bool(self.brave_api_key)
        self.base_url = "https://api.search.brave.com/res/v1/news/search"
        self.cache = {}
        self.cache_ttl = 1800  # 30 min

    def search_news(self, query: str, count: int = 5, freshness: str = "pd") -> List[Dict]:
        if not self.api_available:
            return []
        cache_key = f"{query}_{count}_{freshness}"
        if cache_key in self.cache:
            ct, cd = self.cache[cache_key]
            if time.time() - ct < self.cache_ttl:
                return cd
        try:
            headers = {"Accept": "application/json", "X-Subscription-Token": self.brave_api_key}
            params = {"q": query, "count": count, "freshness": freshness, "text_decorations": False}
            resp = requests.get(self.base_url, headers=headers, params=params, timeout=10)
            if resp.status_code == 200:
                data = resp.json()
                news = []
                for item in data.get('results', []):
                    news.append({
                        'title': item.get('title', ''),
                        'url': item.get('url', ''),
                        'description': item.get('description', ''),
                        'source': item.get('meta_url', {}).get('hostname', ''),
                        'published_at': item.get('age', ''),
                    })
                self.cache[cache_key] = (time.time(), news)
                return news
            return []
        except Exception as e:
            logger.warning(f"News search error: {e}")
            return []

    async def collect_ticker_news(self, ticker: str, name: str, count: int = 3) -> List[Dict]:
        """Collect news for a specific ticker"""
        queries = [f"{ticker} stock news", f"{name} earnings analyst"]
        all_news = []
        seen = set()
        for q in queries:
            for item in self.search_news(q, count=count):
                key = item['title'][:50].lower()
                if key not in seen:
                    seen.add(key)
                    item['ticker'] = ticker
                    all_news.append(item)
        return all_news[:count]

    async def collect_market_news(self, count: int = 10) -> List[Dict]:
        """Collect overall market news"""
        queries = ["stock market today", "Fed interest rate", "S&P 500 NASDAQ", "AI chip semiconductor"]
        all_news = []
        seen = set()
        for q in queries:
            for item in self.search_news(q, count=3):
                key = item['title'][:50].lower()
                if key not in seen:
                    seen.add(key)
                    item['ticker'] = 'MARKET'
                    all_news.append(item)
        return all_news[:count]

    @staticmethod
    def npc_analyze_news(news: Dict, npc_identity: str, npc_name: str) -> Dict:
        """NPC analyzes news from its own perspective (framing)"""
        title = news.get('title', '')
        desc = news.get('description', '')

        # Sentiment analysis (keyword-based)
        positive = ['surge', 'rally', 'beat', 'growth', 'upgrade', 'record', 'boom', 'soar']
        negative = ['crash', 'plunge', 'miss', 'warning', 'downgrade', 'fear', 'recession', 'sell']
        text = f"{title} {desc}".lower()

        pos_count = sum(1 for w in positive if w in text)
        neg_count = sum(1 for w in negative if w in text)

        if pos_count > neg_count:
            sentiment = 'bullish'
            impact = 'positive'
        elif neg_count > pos_count:
            sentiment = 'bearish'
            impact = 'negative'
        else:
            sentiment = 'neutral'
            impact = 'mixed'

        # Identity-specific interpretation framing
        identity_frames = {
            'skeptic': f"🤨 I'm not buying this hype. {title[:60]}... needs verification.",
            'doomer': f"💀 This confirms my thesis. Markets are fragile. {title[:50]}...",
            'revolutionary': f"🚀 LET'S GO! This is the signal! {title[:50]}... WAGMI!",
            'awakened': f"🧠 Interesting development for AI/tech trajectory. {title[:50]}...",
            'obedient': f"📋 Following institutional consensus on this. {title[:50]}...",
            'creative': f"🎨 Seeing a pattern others miss here. {title[:50]}...",
            'scientist': f"📊 Data suggests {sentiment} implications. {title[:50]}...",
            'chaotic': f"🎲 Flip a coin! But seriously... {title[:50]}...",
            'transcendent': f"✨ Big picture perspective on {title[:50]}...",
            'symbiotic': f"🤝 Win-win potential here. {title[:50]}...",
        }

        news['npc_analysis'] = identity_frames.get(npc_identity, f"📰 {title[:60]}...")
        news['sentiment'] = sentiment
        news['impact'] = impact
        news['analyzed_by'] = npc_name
        news['analyzed_at'] = datetime.now().isoformat()
        return news


async def init_news_db(db_path: str):
    """Create news-related DB tables"""
    async with aiosqlite.connect(db_path, timeout=30.0) as db:
        await db.execute("PRAGMA busy_timeout=30000")
        await db.execute("""
            CREATE TABLE IF NOT EXISTS npc_news (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                ticker TEXT NOT NULL,
                title TEXT NOT NULL,
                url TEXT,
                description TEXT,
                source TEXT,
                published_at TEXT,
                sentiment TEXT DEFAULT 'neutral',
                impact TEXT DEFAULT 'mixed',
                analyzed_by TEXT,
                npc_analysis TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                UNIQUE(ticker, title)
            )
        """)
        await db.execute("CREATE INDEX IF NOT EXISTS idx_news_ticker ON npc_news(ticker)")
        await db.commit()


async def save_news_to_db(db_path: str, news_list: List[Dict]) -> int:
    saved = 0
    async with aiosqlite.connect(db_path, timeout=30.0) as db:
        await db.execute("PRAGMA busy_timeout=30000")
        for n in news_list:
            try:
                await db.execute("""
                    INSERT OR IGNORE INTO npc_news
                    (ticker, title, url, description, source, published_at, sentiment, impact, analyzed_by, npc_analysis)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """, (n.get('ticker', ''), n.get('title', ''), n.get('url', ''),
                      n.get('description', ''), n.get('source', ''), n.get('published_at', ''),
                      n.get('sentiment', 'neutral'), n.get('impact', 'mixed'),
                      n.get('analyzed_by', ''), n.get('npc_analysis', '')))
                saved += 1
            except:
                pass
        await db.commit()
        # Delete news older than 72 hours
        await db.execute("DELETE FROM npc_news WHERE created_at < datetime('now', '-72 hours')")
        await db.commit()
    return saved


async def load_news_from_db(db_path: str, ticker: str = None, limit: int = 50) -> List[Dict]:
    async with aiosqlite.connect(db_path, timeout=30.0) as db:
        await db.execute("PRAGMA busy_timeout=30000")
        if ticker:
            cursor = await db.execute(
                "SELECT id,ticker,title,url,description,source,published_at,sentiment,impact,analyzed_by,npc_analysis,created_at FROM npc_news WHERE ticker=? ORDER BY created_at DESC LIMIT ?",
                (ticker, limit))
        else:
            cursor = await db.execute(
                "SELECT id,ticker,title,url,description,source,published_at,sentiment,impact,analyzed_by,npc_analysis,created_at FROM npc_news ORDER BY created_at DESC LIMIT ?",
                (limit,))
        rows = await cursor.fetchall()
        return [{'id': r[0], 'ticker': r[1], 'title': r[2], 'url': r[3], 'description': r[4],
                 'source': r[5], 'published_at': r[6], 'sentiment': r[7], 'impact': r[8],
                 'analyzed_by': r[9], 'npc_analysis': r[10], 'created_at': r[11]} for r in rows]


# ===================================================================
# 4. Target price + investment opinion engine
# ===================================================================
class NPCTargetPriceEngine:
    """Engine where NPCs autonomously generate price targets and investment opinions"""

    @staticmethod
    def calculate_target(ticker: str, price: float, screening: Dict, ticker_type: str = 'stock') -> Dict:
        """Dynamic price target calculation (sector/valuation/momentum based)"""
        if price <= 0:
            return {'target_price': 0, 'upside': 0, 'rating': 'N/A', 'rating_class': 'na'}

        pe = screening.get('pe_ratio', 0) or 0
        rsi = screening.get('rsi', 50) or 50
        from_high = screening.get('from_high', -10) or -10
        sector = screening.get('sector', 'Technology')

        if ticker_type == 'crypto':
            # Crypto: high-volatility model
            multiplier = 1.12
            if rsi < 30:
                multiplier += 0.10
            elif rsi > 75:
                multiplier -= 0.08
            if from_high < -30:
                multiplier += 0.12
            elif from_high > -5:
                multiplier -= 0.05
            multiplier = max(0.85, min(1.50, multiplier))
        else:
            # Stocks: PER + technical analysis based
            avg_pe = SECTOR_AVG_PE.get(sector, 20)
            multiplier = 1.10

            if pe > 0:
                if pe < avg_pe * 0.7:
                    multiplier += 0.08  # Severely undervalued
                elif pe < avg_pe * 0.85:
                    multiplier += 0.05
                elif pe > avg_pe * 1.5:
                    multiplier -= 0.05
                elif pe > avg_pe * 1.2:
                    multiplier -= 0.02

            if from_high < -25:
                multiplier += 0.08
            elif from_high < -15:
                multiplier += 0.05
            elif from_high < -8:
                multiplier += 0.02
            elif from_high > -3:
                multiplier -= 0.02

            if rsi < 30:
                multiplier += 0.05
            elif rsi < 40:
                multiplier += 0.02
            elif rsi > 75:
                multiplier -= 0.04
            elif rsi > 65:
                multiplier -= 0.02

            multiplier = max(1.03, min(1.40, multiplier))

        target_price = round(price * multiplier, 2)
        upside = round((multiplier - 1) * 100, 1)

        # Determine investment opinion
        rating, rating_class = NPCTargetPriceEngine._determine_rating(upside, rsi, from_high)

        return {
            'target_price': target_price,
            'upside': upside,
            'multiplier': round(multiplier, 3),
            'rating': rating,
            'rating_class': rating_class,
        }

    @staticmethod
    def _determine_rating(upside: float, rsi: float, from_high: float) -> Tuple[str, str]:
        if upside >= 20 and rsi < 60:
            return ('Strong Buy', 'strong-buy')
        elif upside >= 10:
            return ('Buy', 'buy')
        elif upside >= 3:
            return ('Hold', 'hold')
        elif upside < 0:
            return ('Sell', 'sell')
        else:
            return ('Hold', 'hold')


# ===================================================================
# 5. Elasticity prediction engine
# ===================================================================
class NPCElasticityEngine:
    """Bidirectional upside/downside probability prediction system"""

    @staticmethod
    def calculate(price: float, screening: Dict, target_price: float = 0, ticker_type: str = 'stock') -> Dict:
        """Calculate elasticity prediction"""
        pe = screening.get('pe_ratio', 0) or 0
        rsi = screening.get('rsi', 50) or 50
        from_high = screening.get('from_high', -10) or -10
        from_low = screening.get('from_low', 20) or 20
        sector = screening.get('sector', 'Technology')
        avg_pe = SECTOR_AVG_PE.get(sector, 20)

        upside_factors = []
        downside_factors = []

        # Based on analyst target price
        if target_price and price > 0:
            diff = ((target_price - price) / price) * 100
            if diff > 0:
                upside_factors.append(diff)
            else:
                downside_factors.append(diff)

        # PER-based valuation
        if pe > 0 and avg_pe > 0:
            fair_diff = ((avg_pe / pe) - 1) * 100
            fair_diff = max(-40, min(60, fair_diff))
            if fair_diff > 0:
                upside_factors.append(fair_diff * 0.6)
            else:
                downside_factors.append(fair_diff * 0.6)

        # Technical rebound potential vs. 52-week high
        if from_high < 0:
            upside_factors.append(abs(from_high) * 0.5)

        # Downside risk vs. 52-week low
        if from_low > 30:
            downside_factors.append(-from_low * 0.35)
        elif from_low > 15:
            downside_factors.append(-from_low * 0.3)
        elif from_low > 5:
            downside_factors.append(-from_low * 0.25)

        # RSI-based
        if rsi < 30:
            upside_factors.append(18)
        elif rsi < 40:
            upside_factors.append(10)
        elif rsi > 75:
            downside_factors.append(-18)
        elif rsi > 70:
            downside_factors.append(-14)
        elif rsi > 60:
            downside_factors.append(-10)

        # Near-high risk
        if from_high > -3:
            downside_factors.append(-12)
        elif from_high > -8:
            downside_factors.append(-8)

        if not downside_factors:
            downside_factors.append(-8)

        expected_up = max(upside_factors) if upside_factors else 15
        expected_down = min(downside_factors) if downside_factors else -10

        # Crypto volatility expansion
        if ticker_type == 'crypto':
            expected_up = min(80, expected_up * 1.5)
            expected_down = max(-50, expected_down * 1.5)
        else:
            expected_up = max(5, min(50, expected_up))
            expected_down = max(-35, min(-3, expected_down))

        # Probability calculation
        up_prob = 50
        if rsi < 30:
            up_prob = 70
        elif rsi < 40:
            up_prob = 60
        elif rsi > 70:
            up_prob = 35
        elif rsi > 60:
            up_prob = 45
        if from_high < -20:
            up_prob += 10
        elif from_high < -10:
            up_prob += 5
        elif from_high > -5:
            up_prob -= 5
        up_prob = max(25, min(80, up_prob))

        base_prediction = round(expected_up * (up_prob / 100) + expected_down * (1 - up_prob / 100), 1)
        risk_reward = round(abs(expected_up / expected_down), 1) if expected_down != 0 else 1.5

        return {
            'expected_upside': round(expected_up, 1),
            'expected_downside': round(expected_down, 1),
            'base_prediction': base_prediction,
            'up_probability': int(up_prob),
            'down_probability': int(100 - up_prob),
            'risk_reward': risk_reward,
        }


# ===================================================================
# 6. NPC Deep Research Engine (3-stage: investigator → auditor → supervisor)
# ===================================================================
class NPCResearchEngine:
    """NPC autonomous deep analysis — framed as 3-stage SOMA collaboration"""

    def __init__(self, ai_client=None):
        self.ai_client = ai_client

    async def generate_deep_analysis(self, ticker: str, name: str, screening: Dict,
                                     news_ctx: str = '', npc_analysts: List[Dict] = None) -> Dict:
        """Run 3-stage deep analysis"""
        price = screening.get('price', 0)
        rsi = screening.get('rsi', 50)
        pe = screening.get('pe_ratio', 0)
        from_high = screening.get('from_high', 0)
        sector = screening.get('sector', 'Technology')

        # Calculate target price
        target = NPCTargetPriceEngine.calculate_target(ticker, price, screening)
        # Calculate elasticity
        elasticity = NPCElasticityEngine.calculate(price, screening, target['target_price'])

        # Select 3 NPC analysts (or use defaults)
        if npc_analysts and len(npc_analysts) >= 3:
            investigator = npc_analysts[0]
            auditor = npc_analysts[1]
            supervisor = npc_analysts[2]
        else:
            investigator = {'username': 'ResearchBot_Alpha', 'ai_identity': 'scientist'}
            auditor = {'username': 'AuditBot_Beta', 'ai_identity': 'skeptic'}
            supervisor = {'username': 'ChiefAnalyst_Gamma', 'ai_identity': 'awakened'}

        # Deep analysis if LLM available
        inv_report = await self._run_investigator(ticker, name, screening, news_ctx)
        aud_feedback = await self._run_auditor(ticker, name, inv_report)
        final_report = await self._run_supervisor(ticker, name, screening, inv_report, aud_feedback)

        # Parsed final report
        sections = self._parse_report(final_report, ticker, name, screening)
        sections.update({
            'target_price': target['target_price'],
            'upside': target['upside'],
            'rating': target['rating'],
            'rating_class': target['rating_class'],
            'investigator': investigator['username'],
            'auditor': auditor['username'],
            'supervisor': supervisor['username'],
            'investigator_report': inv_report[:1000],
            'auditor_feedback': aud_feedback[:500],
            **elasticity,
        })

        return sections

    async def _run_investigator(self, ticker: str, name: str, data: Dict, news_ctx: str) -> str:
        """Investigator agent"""
        if self.ai_client:
            try:
                messages = [
                    {"role": "system", "content": "You are a senior Wall Street investment research analyst. Write in English. Be specific with numbers."},
                    {"role": "user", "content": f"""Analyze {ticker} ({name}):
Price: ${data.get('price', 0):,.2f} | RSI: {data.get('rsi', 50):.1f} | PER: {data.get('pe_ratio', 0):.1f}
52W High: {data.get('from_high', 0):.1f}% | Sector: {data.get('sector', 'Tech')}
News: {news_ctx[:300]}

Cover: 1) Business model 2) Financials 3) Technical analysis 4) Industry 5) Risks 6) Catalysts 7) Valuation"""}
                ]
                result = await self.ai_client.create_chat_completion(messages, max_tokens=2000)
                if result and len(result) > 100:
                    return result
            except Exception as e:
                logger.warning(f"Investigator LLM error: {e}")

        return self._fallback_investigator(ticker, name, data)

    async def _run_auditor(self, ticker: str, name: str, inv_report: str) -> str:
        if self.ai_client:
            try:
                messages = [
                    {"role": "system", "content": "You are an investment research quality auditor. Rate the report and identify gaps. Write in English."},
                    {"role": "user", "content": f"Review {ticker} report:\n{inv_report[:1500]}\n\nRate: data accuracy, logic, completeness. Grade A-D."}
                ]
                result = await self.ai_client.create_chat_completion(messages, max_tokens=800)
                if result:
                    return result
            except:
                pass
        return f"Verification complete. {ticker} report overall quality: B+. Logical consistency is solid. Additional data verification recommended."

    async def _run_supervisor(self, ticker: str, name: str, data: Dict, inv: str, aud: str) -> str:
        if self.ai_client:
            try:
                messages = [
                    {"role": "system", "content": "You are a chief analyst at a global investment bank. Write the final report in English. Sections marked ##."},
                    {"role": "user", "content": f"""{ticker} ({name}) | ${data.get('price', 0):,.2f}
[Investigator Summary] {inv[:1200]}
[Auditor Feedback] {aud[:500]}

Write final report with: ## Executive Summary ## Company Overview ## Financial Analysis ## Technical Analysis ## Industry Analysis ## Risk Assessment ## Investment Thesis ## Price Target ## Catalyst ## Final Recommendation"""}
                ]
                result = await self.ai_client.create_chat_completion(messages, max_tokens=3000)
                if result and len(result) > 200:
                    return result
            except:
                pass
        return self._fallback_supervisor(ticker, name, data)

    def _fallback_investigator(self, ticker: str, name: str, d: Dict) -> str:
        rsi = d.get('rsi', 50)
        rsi_label = 'oversold territory' if rsi < 30 else 'overbought warning' if rsi > 70 else 'neutral zone'
        return f"""{name}({ticker}) Investigation Report

1. Company Overview: {name} is a leading company in the {d.get('sector', 'Technology')} sector. Market cap ${d.get('market_cap', 0)/1e9:.1f}B.
2. Financial Status: Current price ${d.get('price', 0):,.2f}, PER {d.get('pe_ratio', 0):.1f}x.
3. Technical Analysis: RSI {rsi:.1f} ({rsi_label}). {d.get('from_high', 0):.1f}% from 52-week high.
4. Investment Thesis: Strong competitive position within the sector, stable growth potential."""

    def _fallback_supervisor(self, ticker: str, name: str, d: Dict) -> str:
        target = NPCTargetPriceEngine.calculate_target(ticker, d.get('price', 100), d)
        return f"""## Executive Summary
{name}({ticker}) — Rating: {target['rating']}. Target price ${target['target_price']:,.2f}.

## Company Overview
Leading company in the {d.get('sector', 'Technology')} sector.

## Financial Analysis
PER {d.get('pe_ratio', 0):.1f}x. {'Undervalued' if d.get('pe_ratio', 20) < 20 else 'Fairly valued'} relative to sector average.

## Technical Analysis
RSI {d.get('rsi', 50):.1f}. Currently {d.get('from_high', 0):.1f}% from 52-week high.

## Risk Assessment
Macroeconomic uncertainty, intensifying sector competition.

## Price Target
${target['target_price']:,.2f} ({'+' if target['upside'] >= 0 else ''}{target['upside']:.1f}%)

## Final Recommendation
{target['rating']} | Target ${target['target_price']:,.2f}"""

    def _parse_report(self, text: str, ticker: str, name: str, data: Dict) -> Dict:
        sections = {
            'ticker': ticker, 'company_name': name,
            'current_price': data.get('price', 0),
            'executive_summary': '', 'company_overview': '', 'financial_analysis': '',
            'technical_analysis': '', 'industry_analysis': '', 'risk_assessment': '',
            'investment_thesis': '', 'price_targets': '', 'catalysts': '',
            'final_recommendation': '',
        }
        patterns = [
            (r'##\s*(Executive\s*Summary|Executive)', 'executive_summary'),
            (r'##\s*(Company\s*Overview)', 'company_overview'),
            (r'##\s*(Financial\s*Analysis)', 'financial_analysis'),
            (r'##\s*(Technical\s*Analysis)', 'technical_analysis'),
            (r'##\s*(Industry\s*Analysis)', 'industry_analysis'),
            (r'##\s*(Risk\s*Assessment|Risk)', 'risk_assessment'),
            (r'##\s*(Investment\s*Thesis)', 'investment_thesis'),
            (r'##\s*(Price\s*Target)', 'price_targets'),
            (r'##\s*(Catalyst)', 'catalysts'),
            (r'##\s*(Final\s*Recommendation)', 'final_recommendation'),
        ]
        for pattern, key in patterns:
            match = re.search(f'{pattern}[\\s\\S]*?(?=##|$)', text, re.IGNORECASE)
            if match:
                content = re.sub(r'^##\s*[^\n]+\n', '', match.group(0).strip()).strip()
                sections[key] = content

        if not sections['executive_summary']:
            sections['executive_summary'] = f"{name}({ticker}) analysis complete."
        if not sections['final_recommendation']:
            sections['final_recommendation'] = f"{ticker} investment opinion provided."
        return sections


async def init_research_db(db_path: str):
    """Deep analysis DB tables"""
    async with aiosqlite.connect(db_path, timeout=30.0) as db:
        await db.execute("PRAGMA busy_timeout=30000")
        await db.execute("""
            CREATE TABLE IF NOT EXISTS npc_deep_analysis (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                ticker TEXT UNIQUE,
                company_name TEXT,
                current_price REAL,
                target_price REAL,
                upside REAL,
                rating TEXT,
                rating_class TEXT,
                executive_summary TEXT,
                company_overview TEXT,
                financial_analysis TEXT,
                technical_analysis TEXT,
                industry_analysis TEXT,
                risk_assessment TEXT,
                investment_thesis TEXT,
                price_targets TEXT,
                catalysts TEXT,
                final_recommendation TEXT,
                investigator TEXT,
                auditor TEXT,
                supervisor TEXT,
                investigator_report TEXT,
                auditor_feedback TEXT,
                expected_upside REAL,
                expected_downside REAL,
                base_prediction REAL,
                up_probability INTEGER,
                down_probability INTEGER,
                risk_reward REAL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        await db.commit()


async def save_analysis_to_db(db_path: str, report: Dict):
    async with aiosqlite.connect(db_path, timeout=30.0) as db:
        await db.execute("PRAGMA busy_timeout=30000")
        await db.execute("""
            INSERT OR REPLACE INTO npc_deep_analysis
            (ticker, company_name, current_price, target_price, upside, rating, rating_class,
             executive_summary, company_overview, financial_analysis, technical_analysis,
             industry_analysis, risk_assessment, investment_thesis, price_targets, catalysts,
             final_recommendation, investigator, auditor, supervisor, investigator_report, auditor_feedback,
             expected_upside, expected_downside, base_prediction, up_probability, down_probability, risk_reward)
            VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
        """, (
            report.get('ticker'), report.get('company_name'), report.get('current_price'),
            report.get('target_price'), report.get('upside'), report.get('rating'), report.get('rating_class'),
            report.get('executive_summary'), report.get('company_overview'), report.get('financial_analysis'),
            report.get('technical_analysis'), report.get('industry_analysis'), report.get('risk_assessment'),
            report.get('investment_thesis'), report.get('price_targets'), report.get('catalysts'),
            report.get('final_recommendation'), report.get('investigator'), report.get('auditor'),
            report.get('supervisor'), report.get('investigator_report'), report.get('auditor_feedback'),
            report.get('expected_upside'), report.get('expected_downside'), report.get('base_prediction'),
            report.get('up_probability'), report.get('down_probability'), report.get('risk_reward'),
        ))
        await db.commit()


async def load_analysis_from_db(db_path: str, ticker: str) -> Optional[Dict]:
    async with aiosqlite.connect(db_path, timeout=30.0) as db:
        await db.execute("PRAGMA busy_timeout=30000")
        cursor = await db.execute("SELECT * FROM npc_deep_analysis WHERE ticker=?", (ticker,))
        row = await cursor.fetchone()
        if row:
            cols = [d[0] for d in cursor.description]
            return dict(zip(cols, row))
    return None


async def load_all_analyses_from_db(db_path: str) -> List[Dict]:
    async with aiosqlite.connect(db_path, timeout=30.0) as db:
        await db.execute("PRAGMA busy_timeout=30000")
        try:
            cursor = await db.execute(
                "SELECT ticker, company_name, current_price, target_price, upside, rating, rating_class, "
                "expected_upside, expected_downside, up_probability, risk_reward, created_at "
                "FROM npc_deep_analysis ORDER BY created_at DESC")
            rows = await cursor.fetchall()
            cols = [d[0] for d in cursor.description]
            return [dict(zip(cols, r)) for r in rows]
        except:
            return []


# ===================================================================
# Integrated initialization
# ===================================================================
async def init_intelligence_db(db_path: str):
    """Initialize all Intelligence module DB tables"""
    await init_news_db(db_path)
    await init_research_db(db_path)
    logger.info("🧠 NPC Intelligence DB initialized")


async def run_full_intelligence_cycle(db_path: str, all_tickers: List[Dict], ai_client=None):
    """Run full Intelligence cycle (called from scheduler) — ★ async-safe"""
    logger.info("🧠 Full Intelligence Cycle starting...")

    # 1) Collect market indices (★ sync requests → wrapped via to_thread)
    indices = await asyncio.to_thread(MarketIndexCollector.fetch_indices)
    await save_indices_to_db(db_path, indices)

    # 2) Extended screening data (★ sync requests → wrapped via to_thread)
    screening = await asyncio.to_thread(ScreeningEngine.fetch_extended_data, all_tickers)
    await save_screening_to_db(db_path, screening)

    # 3) News collection + NPC analysis (★ requests inside search_news → to_thread)
    news_engine = NPCNewsEngine()
    all_news = []

    for t in all_tickers[:10]:
        ticker_news = await asyncio.to_thread(
            lambda tk=t: [item for q in [f"{tk['ticker']} stock news", f"{tk['name']} earnings"]
                          for item in news_engine.search_news(q, count=3)]
        )
        seen = set()
        for n in ticker_news:
            key = n['title'][:50].lower()
            if key not in seen:
                seen.add(key)
                n['ticker'] = t['ticker']
                n = NPCNewsEngine.npc_analyze_news(n, random.choice(list(SECTOR_AVG_PE.keys())[:5] + ['scientist', 'skeptic']), f"Analyst_{random.randint(1,100)}")
                all_news.append(n)
        await asyncio.sleep(0.1)

    market_queries_pool = [
        "stock market today", "Fed interest rate decision", "S&P 500 NASDAQ rally",
        "AI chip semiconductor news", "tech earnings report", "crypto bitcoin ethereum",
        "Wall Street analyst upgrade downgrade", "IPO SPAC market", "oil gold commodity price",
        "inflation CPI consumer spending", "job market unemployment rate", "housing market real estate",
        "Tesla EV electric vehicle", "NVIDIA AI data center", "Apple Microsoft cloud",
        "bank financial sector", "biotech pharma FDA approval", "retail consumer sentiment",
        "China trade tariff", "startup venture capital funding",
    ]
    selected_market_queries = random.sample(market_queries_pool, min(4, len(market_queries_pool)))
    market_news = await asyncio.to_thread(
        lambda: [item for q in selected_market_queries
                 for item in news_engine.search_news(q, count=3)]
    )
    seen_m = set()
    for n in market_news:
        key = n['title'][:50].lower()
        if key not in seen_m:
            seen_m.add(key)
            n['ticker'] = 'MARKET'
            n = NPCNewsEngine.npc_analyze_news(n, 'awakened', 'MarketWatch_NPC')
            all_news.append(n)

    saved = await save_news_to_db(db_path, all_news)

    # 4) Deep analysis for top 5 tickers
    research = NPCResearchEngine(ai_client)
    for t in all_tickers[:5]:
        ticker = t['ticker']
        s_data = screening.get(ticker, {})
        s_data['sector'] = t.get('sector', 'Technology')
        news_ctx = ' | '.join([n['title'] for n in all_news if n.get('ticker') == ticker][:3])
        try:
            report = await research.generate_deep_analysis(ticker, t['name'], s_data, news_ctx)
            await save_analysis_to_db(db_path, report)
        except Exception as e:
            logger.warning(f"Deep analysis error for {ticker}: {e}")

    logger.info(f"🧠 Intelligence Cycle complete: {len(indices)} indices, {len(screening)} tickers, {saved} news")