from __future__ import annotations import argparse import json import urllib.request from collections import defaultdict from datetime import datetime, timedelta, timezone from decimal import Decimal, InvalidOperation from pathlib import Path from typing import Any IDEAS_PATH = "research/tradingview_community_ideas.jsonl" EVAL_PATH = "research/tradingview_community_idea_evaluations.jsonl" DEFAULT_HORIZONS_HOURS = (24, 168) def _read_jsonl(path: Path) -> list[dict[str, Any]]: if not path.exists(): return [] rows: list[dict[str, Any]] = [] for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): if not line.strip(): continue try: rows.append(json.loads(line)) except json.JSONDecodeError: continue return rows def _decimal(value: Any) -> Decimal | None: try: return Decimal(str(value)) except (InvalidOperation, TypeError, ValueError): return None def _parse_time(value: Any) -> datetime | None: text = str(value or "") if not text: return None if text.endswith("Z"): text = text[:-1] + "+00:00" try: dt = datetime.fromisoformat(text) except ValueError: return None if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.astimezone(timezone.utc) def fetch_hyperliquid_mids(*, timeout: int = 10) -> dict[str, str]: request = urllib.request.Request( "https://api.hyperliquid.xyz/info", data=json.dumps({"type": "allMids"}).encode("utf-8"), headers={"Content-Type": "application/json", "User-Agent": "CTB research-only"}, method="POST", ) with urllib.request.urlopen(request, timeout=timeout) as response: data = json.loads(response.read().decode("utf-8", errors="replace")) return {str(coin).upper(): str(price) for coin, price in data.items()} def _direction_correct(bias: str, return_pct: Decimal) -> bool | None: if bias == "bullish": return return_pct > 0 if bias == "bearish": return return_pct < 0 return None def evaluate_due_ideas( runtime_dir: str | Path = "runtime", *, current_prices: dict[str, str] | None = None, horizons_hours: tuple[int, ...] = DEFAULT_HORIZONS_HOURS, now: datetime | None = None, ) -> dict[str, Any]: runtime = Path(runtime_dir) now = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) prices = current_prices if current_prices is not None else fetch_hyperliquid_mids() ideas = _read_jsonl(runtime / IDEAS_PATH) eval_path = runtime / EVAL_PATH existing = _read_jsonl(eval_path) done = {(str(row.get("url")), int(row.get("horizon_hours") or 0)) for row in existing} new_rows: list[dict[str, Any]] = [] for idea in ideas: url = str(idea.get("url") or "") coin = str(idea.get("coin") or "").upper() bias = str(idea.get("bias") or "neutral") if not url or bias not in {"bullish", "bearish"}: continue entry_price = _decimal(idea.get("price_at_capture_usd")) current_price = _decimal(prices.get(coin)) captured_at = _parse_time(idea.get("timestamp")) if not entry_price or entry_price <= 0 or not current_price or current_price <= 0 or captured_at is None: continue for horizon in horizons_hours: if (url, horizon) in done: continue if now < captured_at + timedelta(hours=horizon): continue ret = (current_price - entry_price) / entry_price * Decimal("100") correct = _direction_correct(bias, ret) row = { "timestamp": now.isoformat(), "source": "tradingview_community_idea_track_record", "url": url, "author": idea.get("author") or "unknown", "coin": coin, "bias": bias, "title": idea.get("title"), "captured_at": captured_at.isoformat(), "horizon_hours": horizon, "entry_price_usd": str(entry_price), "evaluation_price_usd": str(current_price), "return_pct": str(ret.quantize(Decimal("0.01"))), "direction_correct": correct, "research_only": True, "live_order_allowed": False, "mainnet_signed_action": False, } new_rows.append(row) done.add((url, horizon)) if new_rows: eval_path.parent.mkdir(parents=True, exist_ok=True) with eval_path.open("a", encoding="utf-8") as fh: for row in new_rows: fh.write(json.dumps(row, sort_keys=True, ensure_ascii=False) + "\n") return {"status": "ok", "evaluated_new": len(new_rows), "journal": str(eval_path), "summary": summarize_track_record(runtime)} def summarize_track_record(runtime_dir: str | Path = "runtime") -> dict[str, Any]: rows = _read_jsonl(Path(runtime_dir) / EVAL_PATH) buckets: dict[str, dict[str, Any]] = defaultdict(lambda: {"evaluated": 0, "correct": 0}) by_coin: dict[str, dict[str, Any]] = defaultdict(lambda: {"evaluated": 0, "correct": 0}) for row in rows: correct = row.get("direction_correct") is True author = str(row.get("author") or "unknown") coin = str(row.get("coin") or "UNKNOWN").upper() for bucket in (buckets[author], by_coin[coin]): bucket["evaluated"] += 1 if correct: bucket["correct"] += 1 def finish(bucket: dict[str, Any]) -> dict[str, Any]: evaluated = int(bucket["evaluated"]) correct = int(bucket["correct"]) hitrate = Decimal(correct) / Decimal(evaluated) * Decimal("100") if evaluated else Decimal("0") return {"evaluated": evaluated, "correct": correct, "hitrate_pct": str(hitrate.quantize(Decimal("0.01")))} return { "evaluated": len(rows), "by_author": {k: finish(v) for k, v in sorted(buckets.items())}, "by_coin": {k: finish(v) for k, v in sorted(by_coin.items())}, "recommendation": "research_only_until_author_track_record_is_statistically_significant", } def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Evaluate TradingView community idea forward returns as research-only track record.") parser.add_argument("--runtime-dir", default="runtime") parser.add_argument("--summary", action="store_true") args = parser.parse_args(argv) if args.summary: print(json.dumps(summarize_track_record(args.runtime_dir), indent=2, sort_keys=True)) return 0 print(json.dumps(evaluate_due_ideas(args.runtime_dir), indent=2, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())