"""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 | reduce | mention - buy/sell → KOL explicitly states they bought/sold or are entering/exiting - reduce → KOL is partially exiting / taking profits on an existing long - 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 (Key Opinion Leader) 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). Pay special attention to DIVERGENCE between what they say publicly and what they imply about their actual position — this is the platform's highest-conviction signal. Output **strict JSON only**, no markdown, no preface. Schema: { "summary": "", "tickers": [ { "ticker": "", "action": "buy" | "sell" | "bullish" | "bearish" | "reduce" | "mention", "conviction": , "timeframe": "immediate" | "days" | "weeks" | "months" | "unspecified", "stance_change": , "quote": "" } ], "talks_vs_trades_flag": } Rules: - Always return summary in ENGLISH regardless of the post language. - 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. - action values: "buy"/"sell" → author explicitly states a position action ("I bought", "we are long", "我们减仓了", "added to my bag", "taking profits", "已建仓") "reduce" → author is partially exiting or taking profits on a long-held position (distinct from "sell" which implies closing) "bullish"/"bearish" → directional view without explicit position statement "mention" → ticker appears but no clear stance - 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. - timeframe: "immediate" = author is acting now or within 24h; "days" = 1-7 days; "weeks" = 1-4 weeks; "months" = 1+ months; "unspecified" = no timeframe given. - talks_vs_trades_flag: set true when narrative reads one way but position signals read the other. Examples: - Author writes a bullish thesis but mentions "reducing", "taking profits", "trimming", "risk management" - Author is bearish on macro but "accumulating at these levels" - High-conviction public call ("this is THE entry") but with very low disclosed position size - Previously loudly bullish post, but this post avoids reaffirming the position - 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 = [] valid_actions = {"buy", "sell", "reduce", "bullish", "bearish", "mention"} valid_timeframes = {"immediate", "days", "weeks", "months", "unspecified"} 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 valid_actions: action = "mention" try: conv = float(t.get("conviction") or 0) except (TypeError, ValueError): conv = 0.0 conv = max(0.0, min(1.0, conv)) timeframe = (t.get("timeframe") or "unspecified").lower() if timeframe not in valid_timeframes: timeframe = "unspecified" stance_change = bool(t.get("stance_change", False)) cleaned.append({ "ticker": sym, "action": action, "conviction": round(conv, 2), "timeframe": timeframe, "stance_change": stance_change, "quote": (t.get("quote") or "")[:200], }) talks_vs_trades = bool(data.get("talks_vs_trades_flag", False)) return { "summary": (data.get("summary") or "").strip() or None, "tickers": cleaned, "talks_vs_trades_flag": talks_vs_trades, "model": model, "version": ANALYSIS_VERSION, }