add core logic

.gitignore

@@ -0,0 +1,167 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+_docs/_build/
+
+# PyBuilder
+.pybuilder/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# PyCharm
+.idea/
+
+# VS Code
+.vscode/
+
+# Rope
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre
+.pyre/
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.env.*
+.venv
+venv/
+ENV/
+env/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mrc
+.mypy_cache/
+
+# PyInstaller
+*.manifest
+*.spec
+
+# Logs
+logs/
+
+# OS files
+.DS_Store
+Thumbs.db
+
+# Temporary files
+*.tmp
+*.temp
+*.swp
+
+# Caches
+.cache/
+
+# Poetry
+poetry.lock
+
+# Ruff
+.ruff_cache/
+
+# Pyright
+.pyright/
+
+# uv
+.uv/
+
+# Data files
+*.csv
+*.parquet
+*.feather
+
+# Model checkpoints
+*.ckpt

README.md

@@ -9,4 +9,40 @@ license: apache-2.0
 short_description: Market Analyzing Platform - Notifications, Tracking, etc
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+## POC Overview
+This POC implements a market-wide risk radar using:
+- Universe loader for tickers (unofficial lists supported)
+- Price ingestion via `yfinance`
+- Price Delta Calculator (drop by default)
+- Adaptive thresholds, event detection, and alerting
+- In-memory pub/sub topics
+- Process pool for price loading
+
+## Topics
+- `universe.updated`
+- `prices.snapshot`
+- `price.delta.calculated`
+- `event.detected`
+- `alert.ready`
+
+## Run
+```bash
+python -m src.main
+```
+
+## Notes
+- `yfinance` is an unofficial data source and can break or rate-limit.
+- Process pool parallelizes price fetching to speed up ingestion.
+- Universe loader pulls:
+  - US equities from `rreichel3/US-Stock-Symbols` (unofficial)
+  - EU equities from Wikipedia index constituents (FTSE 100, DAX, CAC 40)
+  - Crypto from Binance `exchangeInfo`
+  - Commodities from a curated Yahoo Finance futures list
+
+## Telegram Bot
+Set environment variables:
+```bash
+export TELEGRAM_BOT_TOKEN=...
+export TELEGRAM_CHAT_ID=...
+```
+```

requirements.txt

@@ -0,0 +1,5 @@
+yfinance
+pandas
+numpy
+requests
+lxml

src/core/alerting.py

@@ -0,0 +1,71 @@
+from collections import defaultdict
+from typing import Dict
+import time
+import uuid
+import os
+
+import requests
+
+from .config import MIN_ALERT_GAP_MIN, TELEGRAM_BOT_TOKEN_ENV, TELEGRAM_CHAT_ID_ENV
+
+
+class AlertDeduplicator:
+    def __init__(self) -> None:
+        self._last_ts = defaultdict(lambda: defaultdict(int))
+
+    def allow(self, ticker: str, timeframe: str) -> bool:
+        now = int(time.time())
+        last = self._last_ts[ticker][timeframe]
+        if now - last < MIN_ALERT_GAP_MIN * 60:
+            return False
+        self._last_ts[ticker][timeframe] = now
+        return True
+
+
+def classify(timeframe_minutes: int) -> str:
+    return "flash" if timeframe_minutes <= 30 else "slow"
+
+
+def build_alert(
+    ticker: str,
+    asset_class: str,
+    venue: str,
+    timeframe: str,
+    alert_type: str,
+    delta_pct: float,
+    threshold: float,
+) -> Dict:
+    return {
+        "event_id": str(uuid.uuid4()),
+        "ts": int(time.time() * 1000),
+        "ticker": ticker,
+        "asset_class": asset_class,
+        "venue": venue,
+        "timeframe": timeframe,
+        "alert_type": alert_type,
+        "delta_pct": round(delta_pct, 4),
+        "threshold": round(threshold, 4),
+        "cooldown_sec": MIN_ALERT_GAP_MIN * 60,
+        "destination": "telegram",
+        "message": (
+            f"[{alert_type.upper()}] {ticker} {timeframe} "
+            f"drop {delta_pct:.2f}% (thr {threshold:.2f}%)"
+        ),
+    }
+
+
+class TelegramNotifier:
+    def __init__(self) -> None:
+        self._token = os.getenv(TELEGRAM_BOT_TOKEN_ENV)
+        self._chat_id = os.getenv(TELEGRAM_CHAT_ID_ENV)
+
+    def send(self, alert: Dict) -> None:
+        if not self._token or not self._chat_id:
+            print(alert["message"])
+            return
+        url = f"https://api.telegram.org/bot{self._token}/sendMessage"
+        payload = {"chat_id": self._chat_id, "text": alert["message"]}
+        try:
+            requests.post(url, json=payload, timeout=10)
+        except Exception:
+            print(alert["message"])

src/core/config.py

@@ -0,0 +1,67 @@
+TIMEFRAMES_MIN = {
+    "10m": 10,
+    "30m": 30,
+    "1h": 60,
+    "2h": 120,
+    "3h": 180,
+    "6h": 360,
+    "10h": 600,
+}
+
+K_SIGMA = 3.0
+MIN_ALERT_GAP_MIN = 20
+
+POLL_INTERVAL_SEC = 60
+
+# Process pool for price loading
+PROCESS_POOL_WORKERS = 4
+BATCH_SIZE = 200
+
+# Universe settings
+ASSET_CLASSES = ["equity", "crypto", "commodity"]
+
+# Adaptive threshold parameters
+MIN_WINDOW_POINTS = 5
+
+# Universe sources (no-auth)
+US_TICKER_SOURCES = {
+    "us.nasdaq": "https://raw.githubusercontent.com/rreichel3/US-Stock-Symbols/main/nasdaq/nasdaq_tickers.txt",
+    "us.nyse": "https://raw.githubusercontent.com/rreichel3/US-Stock-Symbols/main/nyse/nyse_tickers.txt",
+    "us.amex": "https://raw.githubusercontent.com/rreichel3/US-Stock-Symbols/main/amex/amex_tickers.txt",
+}
+
+EU_WIKI_INDEX_SOURCES = {
+    "eu.ftse100": {
+        "url": "https://en.wikipedia.org/wiki/FTSE_100_Index",
+        "column": "EPIC",
+        "suffix": ".L",
+    },
+    "eu.dax": {
+        "url": "https://en.wikipedia.org/wiki/DAX",
+        "column": "Ticker symbol",
+        "suffix": ".DE",
+    },
+    "eu.cac40": {
+        "url": "https://en.wikipedia.org/wiki/CAC_40",
+        "column": "Ticker",
+        "suffix": ".PA",
+    },
+}
+
+BINANCE_EXCHANGE_INFO = "https://api.binance.com/api/v3/exchangeInfo"
+
+COMMODITY_TICKERS = [
+    "GC=F",  # Gold
+    "SI=F",  # Silver
+    "CL=F",  # WTI Crude
+    "BZ=F",  # Brent Crude
+    "NG=F",  # Natural Gas
+    "HG=F",  # Copper
+    "ZC=F",  # Corn
+    "ZW=F",  # Wheat
+    "ZS=F",  # Soybeans
+]
+
+# Telegram bot (no auth on load; uses env vars at runtime)
+TELEGRAM_BOT_TOKEN_ENV = "TELEGRAM_BOT_TOKEN"
+TELEGRAM_CHAT_ID_ENV = "TELEGRAM_CHAT_ID"

src/core/event_detector.py

@@ -0,0 +1,29 @@
+from typing import Dict, Optional
+import numpy as np
+
+from .config import K_SIGMA, MIN_WINDOW_POINTS
+
+
+def adaptive_threshold(window) -> Optional[float]:
+    if len(window) < MIN_WINDOW_POINTS:
+        return None
+    prices = np.array(window, dtype=float)
+    returns = np.diff(prices) / prices[:-1]
+    if returns.size == 0:
+        return None
+    sigma = float(np.std(returns))
+    return -K_SIGMA * sigma * 100.0
+
+
+def detect_event(delta_pct: float, threshold: float) -> bool:
+    # Both are negative for drops
+    return delta_pct < threshold
+
+
+def severity(delta_pct: float, threshold: float) -> str:
+    ratio = abs(delta_pct) / max(abs(threshold), 1e-9)
+    if ratio >= 2.0:
+        return "high"
+    if ratio >= 1.3:
+        return "med"
+    return "low"

src/core/main.py

@@ -0,0 +1,115 @@
+import time
+import uuid
+
+from .config import TIMEFRAMES_MIN
+from .pubsub import PubSub
+from .universe_loader import UniverseLoader
+from .price_ingestor import PriceIngestor
+from .price_buffer import PriceBuffer
+from .price_delta import price_delta, velocity_pct_per_min
+from .event_detector import adaptive_threshold, detect_event, severity
+from .alerting import AlertDeduplicator, classify, build_alert, TelegramNotifier
+
+
+TOPICS = {
+    "universe": "universe.updated",
+    "prices": "prices.snapshot",
+    "delta": "price.delta.calculated",
+    "event": "event.detected",
+    "alert": "alert.ready",
+}
+
+
+def run_once():
+    pubsub = PubSub()
+    buffer = PriceBuffer()
+    dedupe = AlertDeduplicator()
+    notifier = TelegramNotifier()
+
+    pubsub.subscribe(TOPICS["alert"], notifier.send)
+
+    loader = UniverseLoader()
+    universe = loader.load()
+
+    for venue_key, tickers in universe.items():
+        if not tickers:
+            continue
+        asset_class = _infer_asset_class(venue_key)
+        venue = venue_key
+        pubsub.publish(TOPICS["universe"], loader.build_event(asset_class, venue, tickers))
+
+        ingestor = PriceIngestor(venue=venue, asset_class=asset_class)
+        price_event = ingestor.fetch_prices(tickers)
+        pubsub.publish(TOPICS["prices"], price_event)
+
+        for p in price_event["prices"]:
+            buffer.update(p["ticker"], p["price"])
+
+        for ticker in tickers:
+            for tf, minutes in TIMEFRAMES_MIN.items():
+                window = buffer.window(ticker, tf)
+                delta = price_delta(window)
+                if not delta:
+                    continue
+                delta_pct = delta["delta_pct"]
+                vel = velocity_pct_per_min(delta_pct, minutes)
+
+                delta_event = {
+                    "event_id": str(uuid.uuid4()),
+                    "ts": int(time.time() * 1000),
+                    "ticker": ticker,
+                    "asset_class": asset_class,
+                    "venue": venue,
+                    "timeframe": tf,
+                    **delta,
+                    "velocity_pct_per_min": vel,
+                    "window_points": len(window),
+                }
+                pubsub.publish(TOPICS["delta"], delta_event)
+
+                thr = adaptive_threshold(window)
+                if thr is None:
+                    continue
+                if detect_event(delta_pct, thr):
+                    if not dedupe.allow(ticker, tf):
+                        continue
+                    event = {
+                        "event_id": str(uuid.uuid4()),
+                        "ts": int(time.time() * 1000),
+                        "ticker": ticker,
+                        "asset_class": asset_class,
+                        "venue": venue,
+                        "timeframe": tf,
+                        "delta_pct": delta_pct,
+                        "threshold": thr,
+                        "threshold_type": "adaptive",
+                        "severity": severity(delta_pct, thr),
+                    }
+                    pubsub.publish(TOPICS["event"], event)
+
+                    alert_type = classify(minutes)
+                    alert = build_alert(
+                        ticker,
+                        asset_class,
+                        venue,
+                        tf,
+                        alert_type,
+                        delta_pct,
+                        thr,
+                    )
+                    pubsub.publish(TOPICS["alert"], alert)
+
+
+
+def _infer_asset_class(venue_key: str) -> str:
+    if "crypto" in venue_key:
+        return "crypto"
+    if "commodities" in venue_key:
+        return "commodity"
+    return "equity"
+
+
+if __name__ == "__main__":
+    while True:
+        run_once()
+        time.sleep(60)

src/core/price_buffer.py

@@ -0,0 +1,23 @@
+from collections import deque
+from typing import Dict, Deque, List
+
+from .config import TIMEFRAMES_MIN
+
+
+class PriceBuffer:
+    def __init__(self) -> None:
+        self._data: Dict[str, Dict[str, Deque[float]]] = {}
+
+    def update(self, ticker: str, price: float) -> None:
+        if ticker not in self._data:
+            self._data[ticker] = {
+                tf: deque(maxlen=window)
+                for tf, window in TIMEFRAMES_MIN.items()
+            }
+        for tf in self._data[ticker]:
+            self._data[ticker][tf].append(price)
+
+    def window(self, ticker: str, tf: str) -> List[float]:
+        if ticker not in self._data or tf not in self._data[ticker]:
+            return []
+        return list(self._data[ticker][tf])

src/core/price_delta.py

@@ -0,0 +1,24 @@
+from typing import Dict, Optional
+
+
+def price_delta(window) -> Optional[Dict[str, float]]:
+    if len(window) < 2:
+        return None
+    start = window[0]
+    now = window[-1]
+    if start == 0:
+        return None
+    delta_abs = now - start
+    delta_pct = (delta_abs / start) * 100.0
+    return {
+        "price_start": float(start),
+        "price_now": float(now),
+        "delta_abs": float(delta_abs),
+        "delta_pct": float(delta_pct),
+    }
+
+
+def velocity_pct_per_min(delta_pct: float, minutes: int) -> float:
+    if minutes <= 0:
+        return 0.0
+    return delta_pct / float(minutes)

src/core/price_ingestor.py

@@ -0,0 +1,83 @@
+from concurrent.futures import ProcessPoolExecutor, as_completed
+from typing import Dict, List
+import time
+import uuid
+
+from .config import BATCH_SIZE, PROCESS_POOL_WORKERS
+
+
+def _fetch_chunk(tickers: List[str]) -> Dict[str, Dict]:
+    import yfinance as yf
+    import pandas as pd
+
+    if not tickers:
+        return {}
+
+    data = yf.download(
+        tickers=" ".join(tickers),
+        period="1d",
+        interval="1m",
+        group_by="ticker",
+        progress=False,
+        threads=False,
+    )
+
+    result: Dict[str, Dict] = {}
+
+    if isinstance(data.columns, pd.MultiIndex):
+        for ticker in tickers:
+            if ticker not in data.columns.get_level_values(0):
+                continue
+            df = data[ticker].dropna()
+            if df.empty:
+                continue
+            price = float(df["Close"].iloc[-1])
+            ts_price = int(df.index[-1].timestamp() * 1000)
+            result[ticker] = {"price": price, "ts_price": ts_price}
+    else:
+        df = data.dropna()
+        if not df.empty:
+            price = float(df["Close"].iloc[-1])
+            ts_price = int(df.index[-1].timestamp() * 1000)
+            result[tickers[0]] = {"price": price, "ts_price": ts_price}
+
+    return result
+
+
+def _chunk_list(items: List[str], size: int) -> List[List[str]]:
+    return [items[i : i + size] for i in range(0, len(items), size)]
+
+
+class PriceIngestor:
+    def __init__(self, venue: str, asset_class: str) -> None:
+        self.venue = venue
+        self.asset_class = asset_class
+
+    def fetch_prices(self, tickers: List[str]) -> Dict:
+        chunks = _chunk_list(tickers, BATCH_SIZE)
+        prices: Dict[str, Dict] = {}
+
+        if PROCESS_POOL_WORKERS <= 1 or len(chunks) <= 1:
+            for chunk in chunks:
+                prices.update(_fetch_chunk(chunk))
+        else:
+            with ProcessPoolExecutor(max_workers=PROCESS_POOL_WORKERS) as pool:
+                futures = [pool.submit(_fetch_chunk, chunk) for chunk in chunks]
+                for f in as_completed(futures):
+                    prices.update(f.result())
+
+        return {
+            "event_id": str(uuid.uuid4()),
+            "ts": int(time.time() * 1000),
+            "asset_class": self.asset_class,
+            "venue": self.venue,
+            "is_delayed": False,
+            "prices": [
+                {
+                    "ticker": t,
+                    "price": p["price"],
+                    "ts_price": p["ts_price"],
+                }
+                for t, p in prices.items()
+            ],
+        }

src/core/pubsub.py

@@ -0,0 +1,14 @@
+from collections import defaultdict
+from typing import Callable, Dict, List, Any
+
+
+class PubSub:
+    def __init__(self) -> None:
+        self._subs: Dict[str, List[Callable[[Any], None]]] = defaultdict(list)
+
+    def subscribe(self, topic: str, handler: Callable[[Any], None]) -> None:
+        self._subs[topic].append(handler)
+
+    def publish(self, topic: str, message: Any) -> None:
+        for handler in self._subs.get(topic, []):
+            handler(message)

src/core/universe_loader.py

@@ -0,0 +1,101 @@
+from typing import Dict, List, Set
+import time
+import uuid
+
+import pandas as pd
+import requests
+
+from .config import (
+    US_TICKER_SOURCES,
+    EU_WIKI_INDEX_SOURCES,
+    BINANCE_EXCHANGE_INFO,
+    COMMODITY_TICKERS,
+)
+
+
+class UniverseLoader:
+    def __init__(self, timeout_sec: int = 20) -> None:
+        self._version = "v1"
+        self._timeout = timeout_sec
+
+    def load(self) -> Dict[str, List[str]]:
+        us = self._load_us_equities()
+        eu = self._load_eu_equities()
+        crypto = self._load_crypto()
+        commodities = sorted(set(COMMODITY_TICKERS))
+
+        return {
+            "us.equities": sorted(us),
+            "eu.equities": sorted(eu),
+            "crypto": sorted(crypto),
+            "commodities": commodities,
+        }
+
+    def build_event(self, asset_class: str, venue: str, tickers: List[str]) -> Dict:
+        return {
+            "event_id": str(uuid.uuid4()),
+            "ts": int(time.time() * 1000),
+            "asset_class": asset_class,
+            "venue": venue,
+            "universe_version": self._version,
+            "tickers": tickers,
+            "changes": {"added": tickers, "removed": []},
+        }
+
+    def _load_us_equities(self) -> Set[str]:
+        tickers: Set[str] = set()
+        for _, url in US_TICKER_SOURCES.items():
+            try:
+                resp = requests.get(url, timeout=self._timeout)
+                resp.raise_for_status()
+                for line in resp.text.splitlines():
+                    t = line.strip().upper()
+                    if not t or t.startswith("#"):
+                        continue
+                    tickers.add(t)
+            except Exception:
+                continue
+        return tickers
+
+    def _load_eu_equities(self) -> Set[str]:
+        tickers: Set[str] = set()
+        for _, cfg in EU_WIKI_INDEX_SOURCES.items():
+            url = cfg["url"]
+            column = cfg["column"]
+            suffix = cfg["suffix"]
+            try:
+                tables = pd.read_html(url)
+            except Exception:
+                continue
+            for table in tables:
+                if column not in table.columns:
+                    continue
+                series = table[column].astype(str).str.strip()
+                for raw in series.tolist():
+                    if not raw or raw == "nan":
+                        continue
+                    sym = raw.split(" ")[0].strip()
+                    if not sym:
+                        continue
+                    if not sym.endswith(suffix):
+                        sym = f"{sym}{suffix}"
+                    tickers.add(sym)
+                break
+        return tickers
+
+    def _load_crypto(self) -> Set[str]:
+        tickers: Set[str] = set()
+        try:
+            resp = requests.get(BINANCE_EXCHANGE_INFO, timeout=self._timeout)
+            resp.raise_for_status()
+            data = resp.json()
+            for item in data.get("symbols", []):
+                if item.get("status") != "TRADING":
+                    continue
+                quote = item.get("quoteAsset")
+                base = item.get("baseAsset")
+                if quote in {"USD", "USDT", "USDC", "BUSD"} and base:
+                    tickers.add(f"{base}-USD")
+        except Exception:
+            return tickers
+        return tickers