run_research.py

#!/usr/bin/env python3
"""Research runner — pipeline principal de investigación Polymarket.

Modos:
  --mode daily     Escanea mercados, calcula discrepancias, genera alertas BUY
  --market <slug>  Analiza un mercado concreto por slug o condition_id
  --calibration    Muestra métricas de calibración del historial de decisiones
  --status         Estado del sistema (decisiones pendientes, cola de research)

Output: JSON a stdout + log a stderr.
Los candidatos BUY se guardan en DB y se imprime el texto de alerta Telegram.
"""

import sys
import os
import json
import argparse
import logging
from datetime import datetime, timezone

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(name)s %(levelname)s: %(message)s",
    stream=sys.stderr,
)
logger = logging.getLogger("polymarket.runner")


def run_daily_scan(
    db_path=None,
    bankroll: float = 100.0,
    max_pages: int = 3,
    mode: str = "paper",
    verbose: bool = False,
    max_llm_calls: int = 15,
) -> dict:
    """Fases 1+2+3+4+5: scan → anchors → LLM research → decide → persist → alert.

    Phase 2 (external anchors) is tried first for every market in the shortlist.
    Markets without an actionable Phase 2 anchor get Phase 3 treatment (Brave +
    Claude CLI), limited to max_llm_calls per run to control token usage.

    Returns:
        dict with keys: status, timestamp, shortlist_size, phase2_candidates,
        phase3_analyzed, buy_decisions, skip_decisions, alerts
    """
    from src.data.market_scanner import MarketScanner
    from src.data.prediction_markets import PredictionMarkets
    from src.data.brave_search import BraveSearchClient
    from src.signals.discrepancy import calculate_discrepancy
    from src.signals.event_researcher import research_market
    from src.signals.event_analyzer import classify_market_category
    from src.execution.decision_engine import ResearchDecisionEngine
    from src.execution.alerter import Alerter
    from src.audit.trade_db import TradeDB
    from src.utils.config import BRAVE_API_KEY

    db = TradeDB(db_path)
    scanner = MarketScanner()
    engine = ResearchDecisionEngine(bankroll=bankroll)
    alerter = Alerter(db=db)
    prediction_client = PredictionMarkets()
    brave_client = BraveSearchClient(api_key=BRAVE_API_KEY) if BRAVE_API_KEY else None

    # --- Fase 1: shortlist (20 markets, macro_quota garantiza política) ---
    logger.info(f"Scanning markets (max_pages={max_pages})...")
    shortlist = scanner.get_shortlist(max_results=20, max_pages=max_pages)
    logger.info(f"Shortlist: {len(shortlist)} markets")

    # Classify semantic topic for anchor routing (politics/sports/crypto/tech/...)
    for market in shortlist:
        if not market.topic_category:
            market.topic_category = classify_market_category(market.question)

    # --- Fase 2: external anchors para cada mercado del shortlist ---
    phase2_candidates: list[tuple] = []   # (market, discrepancy)
    phase3_queue:     list         = []   # markets sin anchor actionable

    for market in shortlist:
        try:
            disc = calculate_discrepancy(
                market=market,
                threshold_pp=8.0,
                friction_pp=4.0,
                prediction_client=prediction_client,
            )
            if disc is not None and disc.actionable:
                phase2_candidates.append((market, disc))
                logger.info(
                    f"Phase2 hit: {market.question[:50]} | "
                    f"{disc.discrepancy_pp:.1f}pp → {disc.suggested_side}"
                )
            else:
                phase3_queue.append(market)
        except Exception as e:
            logger.warning(f"Phase2 failed for {market.question[:40]}: {e}")
            phase3_queue.append(market)

    logger.info(
        f"Phase2: {len(phase2_candidates)} actionable, "
        f"{len(phase3_queue)} queued for Phase3"
    )

    # --- Fase 3: Brave + Claude para TODOS los candidatos con edge ---
    #
    # Priority order for LLM budget:
    #   1. Phase 2 hits (anchor confirms edge) — validated first, most actionable
    #   2. Phase 3 queue (no anchor yet) — pure LLM discovery
    #
    # This ensures every BUY alert has real reasoning, counter-argument, and
    # resolution_risk from Claude. Phase 2 discrepancy is passed as anchor hint.
    #
    # phase3_results: list of (market, research_result, discrepancy_or_None)
    phase3_results: list[tuple] = []   # (market, research_result, discrepancy)
    # Phase 2 candidates that failed LLM (fallback to anchor-only decision)
    phase2_no_llm: list[tuple] = []    # (market, discrepancy)
    llm_calls_used = 0

    if brave_client and max_llm_calls > 0:
        # --- 3a: LLM validation for Phase 2 hits (anchor-confirmed edge) ---
        for market, discrepancy in phase2_candidates:
            if llm_calls_used >= max_llm_calls:
                phase2_no_llm.append((market, discrepancy))
                continue
            try:
                logger.info(f"Phase3 LLM (P2 hit): {market.question[:60]}")
                result = research_market(market, discrepancy=discrepancy, brave_client=brave_client)
                llm_calls_used += 1
                if result.confidence > 0:
                    phase3_results.append((market, result, discrepancy))
                else:
                    # LLM ran but returned 0 confidence (parse error) → fallback
                    logger.warning(f"LLM returned 0 confidence for Phase2 hit: {market.question[:50]}")
                    phase2_no_llm.append((market, discrepancy))
            except Exception as e:
                llm_calls_used += 1
                logger.warning(f"Phase3 LLM failed for Phase2 hit {market.question[:40]}: {e}")
                phase2_no_llm.append((market, discrepancy))

        # --- 3b: LLM discovery for markets without anchor ---
        priority_order = sorted(
            phase3_queue,
            key=lambda m: (0 if getattr(m, "category", "") in {"politics", "economy"} else 1),
        )
        for market in priority_order:
            if llm_calls_used >= max_llm_calls:
                break
            try:
                logger.info(f"Phase3 LLM (no anchor): {market.question[:60]}")
                result = research_market(market, discrepancy=None, brave_client=brave_client)
                llm_calls_used += 1
                if result.confidence > 0:
                    phase3_results.append((market, result, None))
            except Exception as e:
                llm_calls_used += 1
                logger.warning(f"Phase3 failed for {market.question[:40]}: {e}")
    else:
        # No Brave client or LLM disabled — all Phase 2 go anchor-only
        phase2_no_llm = list(phase2_candidates)
        if not brave_client:
            logger.info("BRAVE_API_KEY not set — Phase3 skipped")
        else:
            logger.info("max_llm_calls=0 — Phase3 disabled")

    logger.info(
        f"Phase3: {llm_calls_used} LLM calls, {len(phase3_results)} results "
        f"({len(phase2_no_llm)} Phase2 fallback to anchor-only)"
    )

    # --- Fases 4+5: decidir y persistir todos los candidatos ---
    buy_decisions = []
    skip_count = 0
    alerts = []

    # Phase 2 fallbacks: anchor-only (no LLM — fields will be empty, avoid if possible)
    for market, discrepancy in phase2_no_llm:
        decision = engine.decide(market, discrepancy_result=discrepancy)
        _process_decision(decision, market, db, alerter, mode, verbose,
                          buy_decisions, alerts, logger)
        if decision.verdict == "SKIP":
            skip_count += 1

    if phase2_no_llm:
        logger.info(f"Phase2 anchor-only persisted: {len(phase2_no_llm)} decisions")

    # Phase 3 results: LLM-validated (Phase2 hits + no-anchor discoveries)
    for market, research_result, discrepancy in phase3_results:
        decision = engine.decide(
            market,
            research_result=research_result,
            discrepancy_result=discrepancy,  # passed so engine can cross-check edge
        )
        _process_decision(decision, market, db, alerter, mode, verbose,
                          buy_decisions, alerts, logger)
        if decision.verdict == "SKIP":
            skip_count += 1

    return {
        "status": "ok",
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "shortlist_size": len(shortlist),
        "phase2_candidates": len(phase2_candidates),
        "phase2_llm_validated": len(phase2_candidates) - len(phase2_no_llm),
        "phase2_anchor_only": len(phase2_no_llm),
        "phase3_analyzed": llm_calls_used,
        # candidates_scanned = total markets evaluated (Phase2 + Phase3 LLM runs)
        "candidates_scanned": len(phase2_candidates) + llm_calls_used,
        "buy_decisions": len(buy_decisions),
        "skip_decisions": skip_count,
        "alerts": alerts,
    }


def _process_decision(decision, market, db, alerter, mode, verbose,
                      buy_decisions, alerts, log):
    """Persist ALL decisions (for calibration); alert only BUY signals."""
    category = getattr(market, "topic_category", None) or getattr(market, "category", None)

    # Always persist — SKIPs are essential for calibration (Brier score, win rate)
    db.save_research_decision(decision, category=category, mode=mode)

    if decision.verdict != "SKIP":
        msg, buttons = alerter.format_research_alert(decision, mode=mode)
        alerts.append({
            "decision_id": decision.decision_id,
            "verdict": decision.verdict,
            "market": decision.market_question[:80],
            "edge_net_pp": decision.edge_net_pp,
            "confidence": decision.confidence,
            "amount_suggested": decision.amount_suggested,
            "alert_text": msg,
            "buttons": buttons,
        })
        buy_decisions.append(decision)
        log.info(f"BUY signal: {decision.verdict} — {market.question[:60]}")
    elif verbose:
        log.info(f"SKIP: {market.question[:60]} — {decision.skip_reason}")


def run_market_analysis(
    slug_or_id: str,
    db_path=None,
    bankroll: float = 100.0,
    mode: str = "paper",
) -> dict:
    """Análisis on-demand de un mercado concreto por slug o condition_id.

    Returns:
        dict with decision details or {"status": "error", "error": ...}
    """
    from src.data.market_scanner import MarketScanner
    from src.signals.discrepancy import calculate_discrepancy
    from src.execution.decision_engine import ResearchDecisionEngine
    from src.execution.alerter import Alerter
    from src.audit.trade_db import TradeDB

    db = TradeDB(db_path)
    scanner = MarketScanner()
    engine = ResearchDecisionEngine(bankroll=bankroll)
    alerter = Alerter(db=db)

    # Buscar el mercado
    logger.info(f"Looking up market: {slug_or_id}")
    market = None
    markets = scanner.get_active_markets(limit=200)
    for m in markets:
        if m.slug == slug_or_id or m.condition_id == slug_or_id:
            market = m
            break

    if not market:
        return {"status": "error", "error": f"Market not found: {slug_or_id}"}

    # Enriquecer con CLOB prices
    scanner.enrich_markets_with_clob([market])

    # Calcular discrepancia
    discrepancy = calculate_discrepancy(market)

    # Decidir
    decision = engine.decide(market, discrepancy_result=discrepancy)
    category = getattr(market, "category", None)

    # Always persist (SKIPs too — needed for calibration)
    db.save_research_decision(decision, category=category, mode=mode)

    if decision.verdict != "SKIP":
        msg, buttons = alerter.format_research_alert(decision, mode=mode)
    else:
        msg = f"SKIP: {decision.skip_reason}"
        buttons = []

    return {
        "status": "ok",
        "market": market.question,
        "verdict": decision.verdict,
        "market_price": decision.market_price,
        "our_probability": decision.our_probability,
        "anchor_source": decision.anchor_source,
        "edge_net_pp": decision.edge_net_pp,
        "confidence": decision.confidence,
        "amount_suggested": decision.amount_suggested,
        "skip_reason": decision.skip_reason,
        "alert_text": msg,
        "buttons": buttons,
    }


def run_resolve(db_path=None) -> dict:
    """Resolve pending decisions by checking Polymarket for closed markets."""
    from src.audit.trade_db import TradeDB
    from src.audit.outcome_tracker import OutcomeTracker

    db = TradeDB(db_path)
    tracker = OutcomeTracker(db=db)
    resolved = tracker.check_pending()

    return {
        "status": "ok",
        "resolved_count": len(resolved),
        "resolved": [
            {
                "id": r.get("id"),
                "market": r.get("market_question", "")[:80],
                "outcome": r.get("outcome"),
                "pnl": r.get("pnl_usd"),
            }
            for r in resolved
        ],
    }


def run_calibration(db_path=None) -> dict:
    """Muestra métricas de calibración del historial de decisiones."""
    from src.audit.trade_db import TradeDB
    from src.audit.calibration import summary_metrics

    db = TradeDB(db_path)
    decisions = db.get_research_decisions(limit=500)
    metrics = summary_metrics(decisions)

    return {"status": "ok", "metrics": metrics}


def run_status(db_path=None) -> dict:
    """Estado del sistema: decisiones pendientes y estadísticas."""
    from src.audit.trade_db import TradeDB

    db = TradeDB(db_path)
    pending = db.get_pending_decisions()
    recent = db.get_research_decisions(limit=10)
    stats = db.get_stats()

    return {
        "status": "ok",
        "pending_decisions": len(pending),
        "recent_research_decisions": len(recent),
        "overall_stats": stats,
    }


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Polymarket Research Runner",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Ejemplos:
  python run_research.py --mode daily             # Scan diario completo
  python run_research.py --market will-trump-win  # Análisis on-demand
  python run_research.py --resolve                 # Resolver mercados cerrados
  python run_research.py --calibration            # Métricas históricas
  python run_research.py --status                 # Estado del sistema
        """,
    )
    parser.add_argument(
        "--mode",
        choices=["daily", "paper", "live"],
        default="daily",
        help="Run mode: daily scan, paper, or live (default: daily)",
    )
    parser.add_argument(
        "--market",
        type=str,
        default=None,
        help="Analyze specific market by slug or condition_id",
    )
    parser.add_argument(
        "--resolve",
        action="store_true",
        help="Resolve pending decisions by checking closed markets",
    )
    parser.add_argument(
        "--calibration",
        action="store_true",
        help="Show calibration metrics from decision history",
    )
    parser.add_argument(
        "--status",
        action="store_true",
        help="Show system status (pending decisions, recent activity)",
    )
    parser.add_argument(
        "--bankroll",
        type=float,
        default=100.0,
        help="Total bankroll in USD (default: 100)",
    )
    parser.add_argument(
        "--max-pages",
        type=int,
        default=3,
        help="Max pages to scan (default: 3, ~300 markets)",
    )
    parser.add_argument(
        "--max-llm-calls",
        type=int,
        default=15,
        help="Max Phase3 Claude CLI calls per run (default: 15). "
             "Each call takes ~150s on Mi 9T — set to 5 for cron safety.",
    )
    parser.add_argument(
        "--verbose",
        action="store_true",
        help="Log all SKIP decisions too",
    )
    parser.add_argument(
        "--db",
        type=str,
        default=None,
        help="Path to SQLite DB (default: data/polymarket_bot.db)",
    )
    args = parser.parse_args()

    db_path = args.db

    try:
        if args.status:
            result = run_status(db_path)
        elif args.resolve:
            result = run_resolve(db_path)
        elif args.calibration:
            result = run_calibration(db_path)
        elif args.market:
            result = run_market_analysis(
                args.market,
                db_path=db_path,
                bankroll=args.bankroll,
            )
        else:
            # Default: daily scan
            result = run_daily_scan(
                db_path=db_path,
                bankroll=args.bankroll,
                max_pages=args.max_pages,
                mode="paper" if args.mode == "daily" else args.mode,
                verbose=args.verbose,
                max_llm_calls=args.max_llm_calls,
            )

        print(json.dumps(result, indent=2, default=str))
        sys.exit(0)

    except KeyboardInterrupt:
        logger.info("Interrupted by user")
        sys.exit(1)
    except Exception as e:
        logger.exception(f"Pipeline error: {e}")
        print(json.dumps({"status": "error", "error": str(e)}))
        sys.exit(2)


if __name__ == "__main__":
    main()