| from __future__ import annotations |
|
|
| import hashlib |
| import json |
| import os |
| import threading |
| from dataclasses import dataclass |
| from datetime import UTC, datetime |
| from pathlib import Path |
| from typing import Literal |
|
|
| import pandas as pd |
| from huggingface_hub import download_bucket_files |
|
|
| BUCKET_ID = os.environ.get( |
| "ANALYTICS_BUCKET", |
| "wzsg/polymarket-orderfilled-analytics", |
| ) |
| CACHE_DIR = Path( |
| os.environ.get( |
| "ANALYTICS_CACHE_DIR", |
| "/tmp/polymarket-orderfilled-analytics", |
| ) |
| ) |
| MONTHLY_FILE = "monthly_metrics.parquet" |
| DAILY_FILE = "daily_metrics.parquet" |
| STATE_FILE = "aggregation_state.json" |
|
|
| NUMERIC_COLUMNS = ( |
| "fill_event_count", |
| "unique_transaction_count", |
| "nominal_collateral_volume", |
| "fee_amount", |
| ) |
| Period = Literal["year", "quarter"] |
|
|
|
|
| @dataclass(frozen=True) |
| class Snapshot: |
| monthly: pd.DataFrame |
| daily: pd.DataFrame |
| loaded_at: datetime |
| generated_at: datetime |
| categories: tuple[dict[str, object], ...] |
| category_config_sha256: str | None |
|
|
|
|
| def normalize_frame(frame: pd.DataFrame) -> pd.DataFrame: |
| normalized = frame.copy() |
| if "category" not in normalized: |
| normalized["category"] = "all" |
| for column in NUMERIC_COLUMNS: |
| normalized[column] = pd.to_numeric( |
| normalized[column], |
| errors="coerce", |
| ).fillna(0) |
| return normalized |
|
|
|
|
| def filter_metrics( |
| frame: pd.DataFrame, |
| version: str, |
| market_type: str, |
| category: str = "all", |
| ) -> pd.DataFrame: |
| filtered = frame[ |
| (frame["version"] == version) |
| & (frame["market_type"] == market_type) |
| & (frame["category"] == category) |
| ].copy() |
| order_column = "date" if "date" in filtered.columns else "month" |
| return filtered.sort_values(order_column).reset_index(drop=True) |
|
|
|
|
| def available_market_types( |
| monthly: pd.DataFrame, |
| version: str, |
| ) -> list[str]: |
| order = {"all": 0, "standard": 1, "neg_risk": 2} |
| values = monthly.loc[ |
| monthly["version"] == version, |
| "market_type", |
| ].drop_duplicates() |
| return sorted(values.astype(str).tolist(), key=lambda item: order[item]) |
|
|
|
|
| def available_months( |
| monthly: pd.DataFrame, |
| version: str, |
| market_type: str, |
| category: str = "all", |
| ) -> list[str]: |
| filtered = filter_metrics(monthly, version, market_type, category) |
| return filtered["month"].astype(str).drop_duplicates().tolist() |
|
|
|
|
| def available_categories( |
| monthly: pd.DataFrame, |
| categories: tuple[dict[str, object], ...], |
| version: str, |
| market_type: str, |
| ) -> list[str]: |
| present = set( |
| monthly.loc[ |
| (monthly["version"] == version) |
| & (monthly["market_type"] == market_type), |
| "category", |
| ].astype(str) |
| ) |
| ordered = [ |
| str(item["key"]) |
| for item in sorted(categories, key=lambda item: int(item["order"])) |
| if bool(item["enabled"]) and str(item["key"]) in present |
| ] |
| return ordered or ["all"] |
|
|
|
|
| def _fallback_categories() -> tuple[dict[str, object], ...]: |
| return ( |
| { |
| "key": "all", |
| "order": 0, |
| "enabled": True, |
| "aggregate_only": True, |
| "labels": {"en": "All", "zh": "全部"}, |
| "tag_ids": [], |
| }, |
| ) |
|
|
|
|
| def load_categories_for_state( |
| state: dict[str, object], |
| cache_dir: Path, |
| ) -> tuple[tuple[dict[str, object], ...], str | None]: |
| metadata = state.get("categories") |
| if not isinstance(metadata, dict): |
| return _fallback_categories(), None |
| bucket = metadata.get("bucket") |
| remote_path = metadata.get("path") |
| expected = metadata.get("config_sha256") |
| if not all(isinstance(value, str) and value for value in (bucket, remote_path)): |
| raise ValueError("aggregation_state 分类配置地址无效") |
| if not isinstance(expected, str) or len(expected) != 64: |
| raise ValueError("aggregation_state 分类配置 SHA-256 无效") |
| destination = cache_dir / f"categories-{expected}.json" |
| download_bucket_files( |
| str(bucket), |
| files=[(str(remote_path), destination)], |
| ) |
| raw = destination.read_bytes() |
| actual = hashlib.sha256(raw).hexdigest() |
| if actual != expected: |
| raise RuntimeError( |
| f"分类配置 SHA-256 不匹配：expected={expected}, actual={actual}" |
| ) |
| payload = json.loads(raw) |
| categories = payload.get("categories") |
| if payload.get("schema_version") != 1 or not isinstance(categories, list): |
| raise ValueError("不支持的分类配置 schema") |
| normalized = tuple( |
| item |
| for item in categories |
| if isinstance(item, dict) |
| and isinstance(item.get("key"), str) |
| and isinstance(item.get("order"), int) |
| and isinstance(item.get("enabled"), bool) |
| and isinstance(item.get("labels"), dict) |
| ) |
| if not normalized or not any(item["key"] == "all" for item in normalized): |
| raise ValueError("分类配置缺少 all") |
| return normalized, expected |
|
|
|
|
| def aggregate_period_metrics( |
| monthly: pd.DataFrame, |
| period: Period, |
| ) -> pd.DataFrame: |
| """Roll additive monthly metrics up to UTC calendar periods.""" |
| if period not in ("year", "quarter"): |
| raise ValueError(f"Unsupported period: {period}") |
| if monthly.empty: |
| return pd.DataFrame(columns=["period", *NUMERIC_COLUMNS]) |
|
|
| aggregated = monthly.copy() |
| months = pd.to_datetime( |
| aggregated["month"], |
| format="%Y-%m", |
| errors="raise", |
| ) |
| years = months.dt.year.astype(str) |
| if period == "year": |
| aggregated["period"] = years |
| else: |
| aggregated["period"] = years + "-Q" + months.dt.quarter.astype(str) |
|
|
| return ( |
| aggregated.groupby("period", as_index=False, sort=True)[ |
| list(NUMERIC_COLUMNS) |
| ] |
| .sum() |
| .sort_values("period") |
| .reset_index(drop=True) |
| ) |
|
|
|
|
| class AnalyticsStore: |
| def __init__( |
| self, |
| bucket_id: str = BUCKET_ID, |
| cache_dir: Path = CACHE_DIR, |
| ) -> None: |
| self.bucket_id = bucket_id |
| self.cache_dir = cache_dir |
| self._snapshot: Snapshot | None = None |
| self._lock = threading.Lock() |
|
|
| def refresh(self) -> Snapshot: |
| with self._lock: |
| self.cache_dir.mkdir(parents=True, exist_ok=True) |
| monthly_path = self.cache_dir / MONTHLY_FILE |
| daily_path = self.cache_dir / DAILY_FILE |
| state_path = self.cache_dir / STATE_FILE |
| download_bucket_files( |
| self.bucket_id, |
| files=[ |
| (MONTHLY_FILE, monthly_path), |
| (DAILY_FILE, daily_path), |
| (STATE_FILE, state_path), |
| ], |
| ) |
| state = json.loads(state_path.read_text(encoding="utf-8")) |
| categories, category_sha = load_categories_for_state( |
| state, |
| self.cache_dir, |
| ) |
| monthly = normalize_frame(pd.read_parquet(monthly_path)) |
| daily = normalize_frame(pd.read_parquet(daily_path)) |
| generated = pd.concat( |
| [ |
| pd.to_datetime(monthly["generated_at"], utc=True), |
| pd.to_datetime(daily["generated_at"], utc=True), |
| ] |
| ).max() |
| self._snapshot = Snapshot( |
| monthly=monthly, |
| daily=daily, |
| loaded_at=datetime.now(UTC), |
| generated_at=generated.to_pydatetime(), |
| categories=categories, |
| category_config_sha256=category_sha, |
| ) |
| return self._snapshot |
|
|
| def get(self) -> Snapshot: |
| if self._snapshot is None: |
| return self.refresh() |
| return self._snapshot |
|
|
