"""KOL post → structured signal extractor. Takes a long-form post (Substack essay) or tweet and returns: - summary: one Chinese sentence on what this post is about - tickers: list of {ticker, action, conviction, quote} action ∈ bullish | bearish | buy | sell | mention - buy/sell → KOL explicitly states they bought/sold or are entering/exiting - bullish/bearish → directional view without an explicit position statement - mention → ticker appears but no clear stance (don't flood with these) conviction ∈ 0.0–1.0 - 0.8+ : explicit, repeated, with sizing / timing - 0.5–0.7 : clear view, no commitment - <0.5 : passing reference Quote is the shortest verbatim sentence supporting the call. Uses the same Anthropic client style as analysis.py. Designed to be reused by the Trump-post AI signal module (TODO #4) — same JSON shape, just with different KOL context strings. """ from __future__ import annotations import json import logging from datetime import datetime, timezone from typing import Optional from openai import AsyncOpenAI from app.config import settings logger = logging.getLogger(__name__) ANALYSIS_VERSION = "kol-v1" ANTHROPIC_MODEL = "claude-haiku-4-5-20251001" _anthropic_client = None _openai_client: Optional[AsyncOpenAI] = None def _use_anthropic() -> bool: return bool(settings.anthropic_api_key) def _anth(): global _anthropic_client if _anthropic_client is None: import anthropic as _a _anthropic_client = _a.AsyncAnthropic(api_key=settings.anthropic_api_key) return _anthropic_client def _oai() -> AsyncOpenAI: global _openai_client if _openai_client is None: _openai_client = AsyncOpenAI( api_key=settings.ai_api_key, base_url=settings.ai_base_url, ) return _openai_client SYSTEM_PROMPT = """You are an analyst extracting tradeable signals from crypto KOL posts. The author is a known crypto KOL. Your job: distill what they said and which tokens they are talking about RIGHT NOW (not historical references). Output **strict JSON only**, no markdown, no preface. Schema: { "summary": "", "tickers": [ { "ticker": "", "action": "buy" | "sell" | "bullish" | "bearish" | "mention", "conviction": , "quote": "" } ] } Rules: - If the post is macro commentary, news recap, or sponsored content with no specific token call, return tickers=[] and summary describing the topic. - IGNORE historical price references ("BTC bottomed at $60k earlier this year") — these are context, not current calls. - IGNORE advertising/sponsor sections — look for cues: "sponsor", "partner", "use code", "promo code", "this episode brought to you by", "ad", "广告", "赞助". Skip any ticker only mentioned inside such a section. - buy/sell only when the author states a position action ("I bought", "we are long", "我们减仓了", "added to my bag"). Otherwise use bullish/bearish for directional views, or mention for passing references. - Dedupe per ticker — at most one entry per symbol; pick the strongest action. - Do NOT invent tickers. If you see "$XYZ" but unsure it's a real token, skip it. - conviction: 0.8+ requires explicit + repeated + sized/timed view; 0.5-0.7 for clear directional view without commitment; <0.5 for passing references. - Do not include fiat (USD/CNY/JPY) or stablecoins (USDT/USDC/DAI/FRAX) unless the post's main thesis is about them. """ USER_TEMPLATE = """Today is {today_utc}. KOL handle: {handle} Source: {source} Title: {title} Post body: \"\"\" {body} \"\"\" """ def _truncate(text: str, max_chars: int = 24000) -> str: """Substack essays can be 50K+ chars. Haiku handles it but we cap to control cost. Keep head + tail since conclusions often appear at the end.""" if len(text) <= max_chars: return text head = max_chars * 2 // 3 tail = max_chars - head return text[:head] + "\n\n[...trimmed...]\n\n" + text[-tail:] def _parse_json(raw: str) -> dict: raw = raw.strip() if raw.startswith("```"): # strip fenced code block raw = raw.split("\n", 1)[1] if "\n" in raw else raw if raw.endswith("```"): raw = raw.rsplit("```", 1)[0] raw = raw.strip() # Some models prepend "json" after the fence if raw.startswith("json"): raw = raw[4:].strip() return json.loads(raw) async def extract_kol_signal( *, handle: str, source: str, title: Optional[str], body: str, model: Optional[str] = None, ) -> dict: """Run the extractor. Returns {summary, tickers, model, version}. Returns an empty-but-valid dict on parse/API failure rather than raising — the caller stores the post regardless; an unanalyzed post can be retried. """ today_utc = datetime.now(timezone.utc).strftime("%Y-%m-%d") user = USER_TEMPLATE.format( today_utc=today_utc, handle=handle, source=source, title=title or "", body=_truncate(body), ) use_anth = _use_anthropic() if model is None: # KOL analysis is a daily batch job, not latency-sensitive. Use the # higher-quality `ai_model` (DeepSeek v4 Pro / reasoning) rather than # the live `ai_live_model` (flash) reserved for Trump real-time path. model = ANTHROPIC_MODEL if use_anth else settings.ai_model try: if use_anth: msg = await _anth().messages.create( model=model, max_tokens=1500, temperature=0.1, system=SYSTEM_PROMPT, messages=[{"role": "user", "content": user}], ) raw = (msg.content[0].text if msg.content else "").strip() else: # OpenAI-compatible (DeepSeek). Reasoning models need higher tokens # + no temperature; flash/chat models are fine with both. is_reasoning = any(x in model for x in ("pro", "reasoner", "r1", "think")) kwargs = {"model": model, "messages": [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user}, ], "max_tokens": 4000 if is_reasoning else 1500} if not is_reasoning: kwargs["temperature"] = 0.1 # JSON mode — DeepSeek + OpenAI both support response_format. # Eliminates fenced/preface parse failures. Skipped for reasoning # models (some don't accept response_format alongside reasoning). if not is_reasoning: kwargs["response_format"] = {"type": "json_object"} resp = await _oai().chat.completions.create(**kwargs) raw = (resp.choices[0].message.content or "").strip() data = _parse_json(raw) except Exception as e: logger.warning("kol_analysis extract failed for %s: %s", handle, e) return {"summary": None, "tickers": [], "model": model, "version": ANALYSIS_VERSION, "error": str(e)} # Normalize tickers = data.get("tickers") or [] cleaned = [] for t in tickers: if not isinstance(t, dict): continue sym = (t.get("ticker") or "").strip().upper() if not sym or len(sym) > 12: continue action = (t.get("action") or "mention").lower() if action not in {"buy", "sell", "bullish", "bearish", "mention"}: action = "mention" try: conv = float(t.get("conviction") or 0) except (TypeError, ValueError): conv = 0.0 conv = max(0.0, min(1.0, conv)) cleaned.append({ "ticker": sym, "action": action, "conviction": round(conv, 2), "quote": (t.get("quote") or "")[:200], }) return { "summary": (data.get("summary") or "").strip() or None, "tickers": cleaned, "model": model, "version": ANALYSIS_VERSION, }