import json import logging from fastapi import APIRouter, HTTPException, Query from pydantic import BaseModel from typing import Optional, List, Dict, Any from services.database import ( save_custom_pattern, get_custom_patterns, delete_custom_pattern, toggle_pattern_active, get_config, get_calibration_summary, ) from services.geo_analyzer import PATTERN_TAXONOMY logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/patterns", tags=["patterns"]) class PatternRequest(BaseModel): id: Optional[str] = None name: str description: str triggers: List[str] keywords: List[str] historical_instances: Optional[List[Dict[str, Any]]] = [] suggested_trades: Optional[List[Dict[str, Any]]] = [] asset_class: str expected_move_pct: float probability: float horizon_days: int ai_quality_score: Optional[int] = None ai_evaluation: Optional[Dict[str, Any]] = None source: Optional[str] = "custom" counter_thesis: Optional[str] = None invalidation_trigger: Optional[str] = None invalidation_probability: Optional[float] = None @router.get("/all") def list_all(): """Return all active patterns from DB (builtin + custom).""" return get_custom_patterns() @router.get("/builtin") def list_builtin(): return [p for p in get_custom_patterns() if p.get("source") == "builtin"] @router.get("/custom") def list_custom(): return [p for p in get_custom_patterns() if p.get("source") != "builtin"] @router.post("/custom") def create_pattern(req: PatternRequest): data = req.model_dump() data["source"] = "custom" pat_id = save_custom_pattern(data) return {"id": pat_id, "status": "saved"} @router.put("/custom/{pat_id}") def update_pattern(pat_id: str, req: PatternRequest): data = req.model_dump() data["id"] = pat_id save_custom_pattern(data) return {"id": pat_id, "status": "updated"} class PatternPatchRequest(BaseModel): name: Optional[str] = None description: Optional[str] = None category: Optional[str] = None signal_direction: Optional[str] = None regime_tag: Optional[str] = None @router.patch("/custom/{pat_id}") def patch_pattern(pat_id: str, req: PatternPatchRequest): """Partial update — only provided fields are changed.""" from services.database import get_conn conn = get_conn() if not conn.execute("SELECT id FROM custom_patterns WHERE id=?", (pat_id,)).fetchone(): conn.close() raise HTTPException(404, "Pattern not found") updates = {k: v for k, v in req.model_dump().items() if v is not None} if updates: set_clause = ", ".join(f"{k}=?" for k in updates) conn.execute( f"UPDATE custom_patterns SET {set_clause}, updated_at=datetime('now') WHERE id=?", list(updates.values()) + [pat_id], ) conn.commit() conn.close() return {"id": pat_id, "status": "updated", "fields": list(updates.keys())} @router.delete("/purge-all") def purge_all_patterns(): """Delete ALL custom/backtested patterns from the pattern library.""" from services.database import get_conn conn = get_conn() conn.execute("DELETE FROM custom_patterns WHERE source != 'builtin'") n = conn.total_changes conn.commit() conn.close() return {"deleted": n} @router.delete("/custom/{pat_id}") def delete_pattern(pat_id: str): delete_custom_pattern(pat_id) return {"status": "deleted"} @router.put("/toggle/{pat_id}") def toggle_pattern(pat_id: str): """Enable or disable a pattern (works for builtin and custom).""" new_state = toggle_pattern_active(pat_id) return {"id": pat_id, "is_active": new_state} @router.get("/taxonomy") def get_taxonomy(): """Return the taxonomy tree + all patterns (frontend builds node→pattern mapping).""" patterns = get_custom_patterns() return {"tree": PATTERN_TAXONOMY, "patterns": patterns} @router.get("/by-instrument") def get_by_instrument(ticker: str = Query(..., description="Ticker / underlying to search")): """Return all patterns that have a suggested_trade matching the given ticker.""" ticker_lower = ticker.strip().lower() result = [] for p in get_custom_patterns(): trades = p.get("suggested_trades") or [] matching = [ t for t in trades if ticker_lower in (t.get("underlying") or "").lower() ] if matching: result.append({**p, "matching_trades": matching}) return result # ── Calibration summary ─────────────────────────────────────────────────────── @router.get("/calibration") def get_calibration(): """Per-pattern calibration state: AI estimate vs observed blend.""" return get_calibration_summary() # ── Find Similar ────────────────────────────────────────────────────────────── class FindSimilarRequest(BaseModel): pattern_id: str @router.post("/find-similar") async def find_similar_pattern(req: FindSimilarRequest): """Ask GPT-4o-mini if a library pattern duplicates or counter-scenarios another.""" from openai import AsyncOpenAI openai_key = get_config("openai_api_key") or "" if not openai_key: raise HTTPException(400, "OpenAI API key not configured") library = get_custom_patterns() source = next((p for p in library if p["id"] == req.pattern_id), None) if not source: raise HTTPException(404, "Pattern not found") others = [p for p in library if p["id"] != req.pattern_id] if not others: return { "recommendation": "new_pattern", "match_id": None, "match_name": None, "confidence": 100, "reasoning": "No other patterns in library to compare.", "suggested_regime_tag": source.get("regime_tag") or source.get("category", ""), } def _compact(p: dict) -> dict: trades = p.get("suggested_trades") or [] underlying = trades[0].get("underlying", "") if trades else "" return { "id": p["id"], "name": p["name"], "regime_tag": p.get("regime_tag") or "", "category": p.get("category") or "", "underlying": underlying, "signal_direction": p.get("signal_direction") or "", "description": (p.get("description") or "")[:150], } trades_src = source.get("suggested_trades") or [] underlying_src = trades_src[0].get("underlying", "") if trades_src else "" prompt = f"""You are a quantitative macro pattern analyst. Determine if this library pattern duplicates or counter-scenarios another existing pattern. ## SOURCE PATTERN Name: {source.get('name')} Category: {source.get('category')} Description: {source.get('description')} Underlying: {underlying_src} Signal direction: {source.get('signal_direction')} Regime tag: {source.get('regime_tag') or 'none'} ## EXISTING LIBRARY PATTERNS (excluding source) {json.dumps([_compact(p) for p in others], ensure_ascii=False)} ## DECISION RULES **merge_as_instance**: Same macro REGIME + same INSTRUMENT + same DIRECTION. The source pattern is a redundant instance of an existing one — safe to merge. **counter_scenario**: Same macro REGIME + same INSTRUMENT + OPPOSITE DIRECTION. Both are valid distinct patterns — do NOT merge, just mark the relationship. **new_pattern**: Different regime, instrument, or no meaningful structural match. Keep as-is. CRITICAL: Never merge opposite directions — use counter_scenario instead. Suggest a short generic regime_tag (2-5 words) capturing the macro theme. Return ONLY valid JSON: {{ "recommendation": "merge_as_instance" | "counter_scenario" | "new_pattern", "match_id": "matched pattern id or null", "match_name": "matched pattern name or null", "confidence": 0-100, "reasoning": "2-3 sentences explaining the decision", "suggested_regime_tag": "short generic label" }}""" client = AsyncOpenAI(api_key=openai_key) resp = await client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], temperature=0.1, response_format={"type": "json_object"}, ) raw = resp.choices[0].message.content or "{}" try: return json.loads(raw) except Exception: raise HTTPException(500, f"AI returned non-JSON: {raw[:200]}") # ── Merge Patterns ──────────────────────────────────────────────────────────── class MergeRequest(BaseModel): keep_id: str discard_id: str @router.post("/merge") def merge_patterns(req: MergeRequest): """Merge discard_id into keep_id. - Migrates all child-table rows (scores, trades, traces, logs, embeddings) - Merges historical_instances (union, dedup by date+event) - Sums backtest_hits + backtest_runs_count - Deletes the discarded pattern """ from services.database import get_conn if req.keep_id == req.discard_id: raise HTTPException(400, "keep_id and discard_id must be different") conn = get_conn() try: keep_row = conn.execute("SELECT * FROM custom_patterns WHERE id=?", (req.keep_id,)).fetchone() discard_row = conn.execute("SELECT * FROM custom_patterns WHERE id=?", (req.discard_id,)).fetchone() if not keep_row: raise HTTPException(404, f"Pattern {req.keep_id} not found") if not discard_row: raise HTTPException(404, f"Pattern {req.discard_id} not found") keep = dict(keep_row) discard = dict(discard_row) # ── Merge historical_instances ──────────────────────────────────────── def _load_instances(raw) -> list: if not raw: return [] try: v = json.loads(raw) return v if isinstance(v, list) else [] except Exception: return [] keep_inst = _load_instances(keep.get("historical_instances")) discard_inst = _load_instances(discard.get("historical_instances")) seen = {(i.get("date", ""), i.get("event", "")) for i in keep_inst} for inst in discard_inst: key = (inst.get("date", ""), inst.get("event", "")) if key not in seen: keep_inst.append(inst) seen.add(key) keep_inst.sort(key=lambda x: x.get("date", ""), reverse=True) # ── Merge numeric counters ──────────────────────────────────────────── merged_hits = (keep.get("backtest_hits") or 0) + (discard.get("backtest_hits") or 0) merged_runs = (keep.get("backtest_runs_count") or 0) + (discard.get("backtest_runs_count") or 0) # Prefer keep's regime_tag; fall back to discard's merged_regime = keep.get("regime_tag") or discard.get("regime_tag") # ── Remap child tables ──────────────────────────────────────────────── _CHILD_TABLES = [ "pattern_score_history", "trade_entry_prices", "ai_reasoning_traces", "ai_call_logs", "skipped_trades", ] for tbl in _CHILD_TABLES: try: conn.execute( f"UPDATE {tbl} SET pattern_id=? WHERE pattern_id=?", (req.keep_id, req.discard_id), ) except Exception as e: logger.warning(f"[merge] {tbl} remap failed (may not exist): {e}") # pattern_embeddings: unique constraint on pattern_id — delete discard's embedding # (keep's embedding stays; it'll be regenerated on next cycle if needed) try: conn.execute("DELETE FROM pattern_embeddings WHERE pattern_id=?", (req.discard_id,)) except Exception: pass # ── Update keep pattern ─────────────────────────────────────────────── conn.execute( """UPDATE custom_patterns SET historical_instances=?, backtest_hits=?, backtest_runs_count=?, regime_tag=?, updated_at=datetime('now') WHERE id=?""", ( json.dumps(keep_inst, ensure_ascii=False), merged_hits, merged_runs, merged_regime, req.keep_id, ), ) # ── Delete discarded pattern ────────────────────────────────────────── conn.execute("DELETE FROM custom_patterns WHERE id=?", (req.discard_id,)) conn.commit() logger.info(f"[Merge] Kept {req.keep_id}, discarded {req.discard_id} — {len(keep_inst)} instances, {merged_hits}/{merged_runs} backtest hits") updated = conn.execute("SELECT * FROM custom_patterns WHERE id=?", (req.keep_id,)).fetchone() return {"status": "merged", "kept_id": req.keep_id, "discarded_id": req.discard_id, "merged_instances": len(keep_inst), "pattern": dict(updated)} finally: conn.close()