import streamlit as st from pathlib import Path import json import requests from datetime import datetime, timezone, timedelta import numpy as np from zoneinfo import ZoneInfo from datetime import datetime, timedelta from config import SESSION @st.cache_data(ttl=60) def fetch_user_trackings(): url = "https://xtracker.polymarket.com/api/users/elonmusk" # uses the xtracker on polymarket logic needs to be as it is described here, never replace try: r = SESSION.get(url, timeout=12) r.raise_for_status() data = r.json() if data.get("success"): return data["data"]["trackings"] return [] except Exception as e: st.warning(f"XTracker users failed: {e}") return [] buckets = [] for m in event.get("markets", [event]): q = m.get("question") or "" if not q: continue # Normalized Label + Range (aus vorheriger Anpassung) clean_label, low, high = normalize_bucket(q) mid = midpoint(low, high) # === NEU: Token-ID extrahieren === # to get clob_tokens = m.get("clobTokenIds") if isinstance(clob_tokens, str): try: clob_tokens = json.loads(clob_tokens) except: clob_tokens = [] token_id = clob_tokens[0] if clob_tokens else None try: outcome_str = m.get("outcomePrices", '["0","1"]') prices = json.loads(outcome_str) if isinstance(outcome_str, str) else (outcome_str or [0]) yes_price = float(prices[0]) except: yes_price = 0.0 buckets.append({ "label": clean_label, "low": low, "high": high, "mid": mid, "market_prob": yes_price, "token_id": token_id, # ← NEU & WICHTIG für Orders }) def find_tracking_id_by_slug(slug: str, trackings: list): ## this logic gets the id-string needed to identifie the desired event on polymarkets xtracker, this is mandatory and not to replace slug_norm = slug.lower().replace("-", " ").replace("of", "").strip() for t in trackings: title_norm = t.get("title", "").lower().replace("#", "").replace("?", "").replace("-", " ").strip() if all(word in title_norm for word in slug_norm.split() if word.isdigit() or len(word) > 3): return t["id"] return None @st.cache_data(ttl=45) def fetch_xtracker_data(tracking_id: str): ## using the tracking_id looked up with find_tracking_id_by_slug(slug,trackings) fetches data and returns it. this is mandatory and not to replace url = f"https://xtracker.polymarket.com/api/trackings/{tracking_id}?includeStats=true" try: r = SESSION.get(url, timeout=12) r.raise_for_status() data = r.json() return data.get("data") if data.get("success") else None except: return None @st.cache_data(ttl=300) def fetch_xtracker_cached(): ## data from xtracker is fetched and cached in a json file this is also mandatory and not to replace cache_file = Path("xtracker_cache.json") # Load cache if cache_file.exists(): try: with open(cache_file) as f: cache = json.load(f) tweets = cache.get("tweets", []) last_ts = cache.get("last_ts") except: tweets, last_ts = [], None else: tweets, last_ts = [], None # Determine start time for incremental fetch start_iso = last_ts if last_ts else (datetime.now(timezone.utc) - timedelta(days=30)).strftime("%Y-%m-%dT%H:%M:%S.000Z") end_iso = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.000Z") url = f"https://xtracker.polymarket.com/api/users/elonmusk/posts?startDate={start_iso}&endDate={end_iso}" try: r = requests.get(url, timeout=15) r.raise_for_status() new_posts = r.json().get("data", []) # as you corrected if new_posts: # DEDUPLICATION by tweet ID (this is the real fix) existing_ids = {p.get("id") or p.get("tweetId") for p in tweets if p.get("id") or p.get("tweetId")} truly_new = [p for p in new_posts if (p.get("id") or p.get("tweetId")) not in existing_ids] if truly_new: tweets.extend(truly_new) # Update last_ts from the newest real tweet last_ts = max((p.get("createdAt") for p in truly_new), default=last_ts, key=str) # Save clean cache with open(cache_file, "w") as f: json.dump({"tweets": tweets, "last_ts": last_ts}, f) print(f"✅ Added {len(truly_new)} truly new tweets (cache now: {len(tweets)})") # quiet console only else: print("✅ No new tweets since last update") # quiet console only except Exception as e: st.warning(f"XTracker fetch failed: {e}") return tweets def get_18h_projections(tracker_data: dict): if not tracker_data: return None stats = tracker_data["stats"] current = stats["total"] end_dt = datetime.fromisoformat(tracker_data["endDate"].replace("Z", "+00:00")) remaining = max((end_dt - datetime.now(timezone.utc)).total_seconds() / 3600, 0.1) hourly = [e["count"] for e in stats["daily"]] h30 = float(np.mean(hourly[-720:])) if len(hourly) >= 720 else 2.3 h7 = float(np.mean(hourly[-168:])) if len(hourly) >= 168 else h30 h24 = float(np.mean(hourly[-24:])) if len(hourly) >= 24 else h7 return { "proj_30d": round(current + h30 * remaining, 1), "proj_7d": round(current + h7 * remaining, 1), "proj_24h": round(current + h24 * remaining, 1), "current_in_period": int(current), "remaining_hours": round(remaining, 1) } def get_pace(days: float) -> float: """Calculate average tweets per hour over last N days. Moved here so DataHub + trading bot can import it cleanly.""" cutoff = datetime.now(ZoneInfo("Europe/Berlin")) - timedelta(days=days) tweets = fetch_xtracker_cached() count = len([ p for p in tweets if datetime.fromisoformat(p.get("createdAt", "").replace("Z", "+00:00")) >= cutoff ]) hours = (datetime.now(ZoneInfo("Europe/Berlin")) - cutoff).total_seconds() / 3600 return float(count) / float(hours) if hours > 0 else 0.0