Blame for scripts/python/ingest_existing_feeds.py

#!/usr/bin/env python3
"""

Ingest Existing Feeds into Global History DB

=============================================

Wires the existing Aureon feed modules (CoinGecko, Global Financial Feed,

News Sentiment, Glassnode On-Chain, Coinbase Historical, Macro Intelligence)

into the unified SQLite global history database.



Each feed is independently try/except-wrapped so one failure never blocks

the rest.



Usage:

    python ingest_existing_feeds.py

    python ingest_existing_feeds.py --feeds coingecko,coinbase

    python ingest_existing_feeds.py --feeds all --coinbase-pairs BTC-USD,ETH-USD

"""

from __future__ import annotations

import sys
from pathlib import Path

# ---------------------------------------------------------------------------
# Repo-root sys.path setup (same pattern as sync_global_history_db.py)
# ---------------------------------------------------------------------------
_REPO_ROOT = Path(__file__).resolve().parents[2]
for _p in (
    _REPO_ROOT,
    _REPO_ROOT / "aureon" / "core",
    _REPO_ROOT / "aureon" / "exchanges",
    _REPO_ROOT / "aureon" / "data_feeds",
    _REPO_ROOT / "aureon" / "intelligence",
    _REPO_ROOT / "aureon" / "monitors",
):
    _sp = str(_p)
    if _sp not in sys.path:
        sys.path.insert(0, _sp)

try:
    from aureon_baton_link import link_system as _baton_link
    _baton_link(__name__)
except Exception:
    pass

import argparse
import builtins
import io
import json
import os
import time
from datetime import datetime, timezone

# ---------------------------------------------------------------------------
# DB module (always required)
# ---------------------------------------------------------------------------
import aureon_global_history_db as db

_STDOUT_FD: int | None = None
try:
    if hasattr(sys.__stdout__, "fileno"):
        _STDOUT_FD = os.dup(sys.__stdout__.fileno())
except Exception:
    _STDOUT_FD = None


def _restore_stdout() -> None:
    """Recover after legacy feed modules replace and close stdout."""
    if not getattr(sys.stdout, "closed", False):
        return
    if _STDOUT_FD is None:
        sys.stdout = sys.__stdout__
        return
    try:
        raw = os.fdopen(os.dup(_STDOUT_FD), "wb", closefd=True)
        sys.stdout = io.TextIOWrapper(raw, encoding="utf-8", errors="replace", line_buffering=True)
    except Exception:
        sys.stdout = sys.__stdout__


def print(*args, **kwargs):  # type: ignore[override]
    _restore_stdout()
    return builtins.print(*args, **kwargs)

# ---------------------------------------------------------------------------
# Defaults
# ---------------------------------------------------------------------------
DEFAULT_COINGECKO_COINS = [
    "bitcoin", "ethereum", "ripple", "solana", "cardano", "dogecoin",
    "polkadot", "avalanche-2", "chainlink", "matic-network", "litecoin",
    "uniswap", "cosmos", "near", "stellar",
]

DEFAULT_COINBASE_PAIRS = [
    "BTC-USD", "ETH-USD", "XRP-USD", "SOL-USD", "ADA-USD", "DOGE-USD",
]

DEFAULT_NEWS_KEYWORDS = [
    "market", "stocks", "crypto", "forex", "economy", "inflation",
    "fed", "geopolitical",
]

ALL_FEEDS = ["coingecko", "macro", "news", "glassnode", "coinbase", "macro_intel"]

# Macro snapshot fields we want to persist as individual sentiment rows.
MACRO_SNAPSHOT_FIELDS = [
    "crypto_fear_greed", "crypto_fg_classification",
    "vix", "vix_change",
    "dxy", "dxy_change",
    "us_10y_yield", "us_2y_yield", "yield_curve_inversion",
    "spx", "spx_change",
    "gold", "gold_change",
    "oil", "oil_change",
    "total_crypto_mcap", "btc_dominance",
    "eur_usd", "gbp_usd", "usd_jpy",
    "risk_on_off", "market_regime",
]


def _now_ms() -> int:
    return int(time.time() * 1000)


def _ts_id(source: str, ts_ms: int, suffix: str = "") -> str:
    """Generate a deterministic unique ID for deduplication."""
    base = f"{source}:{ts_ms}"
    return f"{base}:{suffix}" if suffix else base


# ===================================================================
# 1. CoinGecko
# ===================================================================
def ingest_coingecko(conn, coins: list[str]) -> dict:
    """Fetch current CoinGecko prices and store as point-in-time market bars."""
    try:
        from coingecko_price_feeder import fetch_coingecko_prices
    except ImportError as exc:
        return {"status": "skipped", "reason": f"import failed: {exc}"}

    _restore_stdout()
    print("\n[coingecko] Fetching prices ...")
    prices = fetch_coingecko_prices(coins)
    if not prices:
        return {"status": "ok", "inserted": 0, "note": "no data returned"}

    now = _now_ms()
    inserted = 0
    for coin_id, data in prices.items():
        usd_price = data.get("usd")
        if usd_price is None:
            continue
        bar = {
            "provider": "coingecko",
            "venue": "coingecko",
            "symbol_id": f"coingecko:{coin_id}",
            "symbol": coin_id,
            "period_id": "snapshot",
            "time_start_ms": now,
            "time_end_ms": now,
            "open": usd_price,
            "high": usd_price,
            "low": usd_price,
            "close": usd_price,
            "volume": data.get("usd_24h_vol"),
            "trades_count": None,
            "ingested_at_ms": now,
            "raw_json": json.dumps(data),
        }
        if db.insert_market_bar(conn, bar):
            inserted += 1
    conn.commit()
    db.set_watermark_ms(conn, "coingecko", now)
    conn.commit()
    print(f"[coingecko] Inserted {inserted} market bars for {len(prices)} coins.")
    return {"status": "ok", "inserted": inserted, "coins": len(prices)}


# ===================================================================
# 2. Global Financial Feed (macro snapshot)
# ===================================================================
def ingest_macro(conn) -> dict:
    """Take a GlobalFinancialFeed snapshot and store each indicator."""
    try:
        from global_financial_feed import GlobalFinancialFeed
    except ImportError as exc:
        return {"status": "skipped", "reason": f"import failed: {exc}"}

    print("\n[macro] Fetching global financial snapshot ...")
    feed = GlobalFinancialFeed()
    snap = feed.get_snapshot()
    snap_dict = snap.to_dict() if hasattr(snap, "to_dict") else {}

    now = _now_ms()
    inserted = 0
    for field_name in MACRO_SNAPSHOT_FIELDS:
        raw_value = getattr(snap, field_name, None)
        if raw_value is None:
            raw_value = snap_dict.get(field_name)
        if raw_value is None:
            continue

        # Coerce booleans / strings to a numeric value where possible
        numeric_value = None
        label = None
        if isinstance(raw_value, bool):
            numeric_value = 1.0 if raw_value else 0.0
            label = str(raw_value)
        elif isinstance(raw_value, (int, float)):
            numeric_value = float(raw_value)
        elif isinstance(raw_value, str):
            label = raw_value

        row = {
            "source": "global_financial_feed",
            "sentiment_id": _ts_id("gff", now, field_name),
            "sentiment_type": field_name,
            "symbol": None,
            "value": numeric_value,
            "label": label,
            "ts_ms": now,
            "ingested_at_ms": now,
            "raw_json": json.dumps({"field": field_name, "raw": raw_value}),
        }
        if db.insert_sentiment(conn, row):
            inserted += 1
    conn.commit()
    db.set_watermark_ms(conn, "global_financial_feed", now)
    conn.commit()
    print(f"[macro] Inserted {inserted} sentiment rows from macro snapshot.")
    return {"status": "ok", "inserted": inserted}


# ===================================================================
# 3. News Sentiment
# ===================================================================
def ingest_news(conn, keywords: list[str]) -> dict:
    """Fetch recent news articles, store each + aggregate as sentiment rows."""
    try:
        from aureon_news_feed import NewsFeed, NewsFeedConfig
    except ImportError as exc:
        return {"status": "skipped", "reason": f"import failed: {exc}"}

    api_key = os.environ.get("WORLD_NEWS_API_KEY", "")
    if not api_key:
        return {"status": "skipped", "reason": "WORLD_NEWS_API_KEY not set"}

    _restore_stdout()
    print("\n[news] Fetching recent articles ...")
    import asyncio

    config = NewsFeedConfig(api_key=api_key)
    feed = NewsFeed(config)
    now = _now_ms()
    inserted_articles = 0
    inserted_agg = 0

    async def _fetch_all():
        all_articles = []
        for kw in keywords:
            try:
                arts = await feed.search_news(text=kw, number=10)
                all_articles.extend(arts)
            except Exception as exc:
                print(f"  [news] search_news('{kw}') failed: {exc}")
        return all_articles

    articles = asyncio.run(_fetch_all())

    # Deduplicate by article ID
    seen_ids: set = set()
    unique_articles = []
    for art in articles:
        if art.id not in seen_ids:
            seen_ids.add(art.id)
            unique_articles.append(art)

    # Store each article as a sentiment row
    for art in unique_articles:
        pub_ms = now
        try:
            dt = datetime.strptime(art.publish_date, "%Y-%m-%d %H:%M:%S")
            pub_ms = int(dt.replace(tzinfo=timezone.utc).timestamp() * 1000)
        except Exception:
            pass

        row = {
            "source": "news_feed",
            "sentiment_id": _ts_id("news_article", pub_ms, str(art.id)),
            "sentiment_type": "news_article",
            "symbol": art.category,
            "value": art.sentiment,
            "label": art.sentiment_label,
            "ts_ms": pub_ms,
            "ingested_at_ms": now,
            "raw_json": json.dumps({
                "id": art.id,
                "title": art.title,
                "url": art.url,
                "summary": art.summary[:500] if art.summary else None,
                "category": art.category,
                "sentiment": art.sentiment,
            }),
        }
        if db.insert_sentiment(conn, row):
            inserted_articles += 1

    # Store aggregate sentiment
    if unique_articles:
        agg = feed.analyze_sentiment_aggregate(unique_articles)
        signals = feed.extract_market_signals(unique_articles)
        agg_row = {
            "source": "news_feed",
            "sentiment_id": _ts_id("news_agg", now),
            "sentiment_type": "news_aggregate",
            "symbol": None,
            "value": agg.get("average_sentiment"),
            "label": f"bullish_ratio={agg.get('bullish_ratio')},bearish_ratio={agg.get('bearish_ratio')}",
            "ts_ms": now,
            "ingested_at_ms": now,
            "raw_json": json.dumps({"aggregate": agg, "signals": signals}),
        }
        if db.insert_sentiment(conn, agg_row):
            inserted_agg += 1

    conn.commit()
    db.set_watermark_ms(conn, "news_feed", now)
    conn.commit()
    print(f"[news] Inserted {inserted_articles} article rows + {inserted_agg} aggregate rows "
          f"({len(unique_articles)} unique articles).")
    return {"status": "ok", "articles": inserted_articles, "aggregates": inserted_agg}


# ===================================================================
# 4. Glassnode On-Chain
# ===================================================================
def ingest_glassnode(conn, hours_back: int) -> dict:
    """Fetch whale movements and exchange flows, store as onchain_metric rows."""
    try:
        from glassnode_client import GlassnodeClient
    except ImportError as exc:
        return {"status": "skipped", "reason": f"import failed: {exc}"}

    api_key = os.environ.get("GLASSNODE_API_KEY", "")
    if not api_key:
        return {"status": "skipped", "reason": "GLASSNODE_API_KEY not set"}

    _restore_stdout()
    print(f"\n[glassnode] Fetching on-chain data (last {hours_back}h) ...")
    client = GlassnodeClient()
    now = _now_ms()
    inserted = 0

    # -- Whale movements --
    try:
        whales = client.detect_whale_movements(min_usd_value=100_000, hours_back=hours_back)
        for wm in whales:
            ts = int(wm.timestamp * 1000) if wm.timestamp < 1e12 else int(wm.timestamp)
            row = {
                "provider": "glassnode",
                "metric_id": _ts_id("whale", ts, wm.asset),
                "metric_name": "whale_movement",
                "asset": wm.asset,
                "value": wm.usd_value,
                "ts_ms": ts,
                "ingested_at_ms": now,
                "raw_json": json.dumps({
                    "amount": wm.amount,
                    "usd_value": wm.usd_value,
                    "asset": wm.asset,
                }),
            }
            if db.insert_onchain_metric(conn, row):
                inserted += 1
    except Exception as exc:
        print(f"  [glassnode] detect_whale_movements failed: {exc}")

    # -- Exchange flows --
    try:
        flows = client.get_exchange_flows(hours_back=hours_back)
        for ef in flows:
            ts = int(ef.timestamp * 1000) if ef.timestamp < 1e12 else int(ef.timestamp)
            row = {
                "provider": "glassnode",
                "metric_id": _ts_id("exflow", ts, ef.asset),
                "metric_name": "exchange_flow",
                "asset": ef.asset,
                "value": ef.net_flow_usd,
                "ts_ms": ts,
                "ingested_at_ms": now,
                "raw_json": json.dumps({
                    "inflow_usd": ef.inflow_usd,
                    "outflow_usd": ef.outflow_usd,
                    "net_flow_usd": ef.net_flow_usd,
                    "asset": ef.asset,
                    "exchange": getattr(ef, "exchange", None),
                }),
            }
            if db.insert_onchain_metric(conn, row):
                inserted += 1
    except Exception as exc:
        print(f"  [glassnode] get_exchange_flows failed: {exc}")

    # -- Intelligence summary (store as single snapshot metric) --
    try:
        summary = client.get_whale_intelligence_summary()
        row = {
            "provider": "glassnode",
            "metric_id": _ts_id("intel_summary", now),
            "metric_name": "whale_intelligence_summary",
            "asset": "ALL",
            "value": summary.get("large_tx_volume_24h"),
            "ts_ms": now,
            "ingested_at_ms": now,
            "raw_json": json.dumps(summary),
        }
        if db.insert_onchain_metric(conn, row):
            inserted += 1
    except Exception as exc:
        print(f"  [glassnode] get_whale_intelligence_summary failed: {exc}")

    conn.commit()
    db.set_watermark_ms(conn, "glassnode", now)
    conn.commit()
    print(f"[glassnode] Inserted {inserted} onchain_metric rows.")
    return {"status": "ok", "inserted": inserted}


# ===================================================================
# 5. Coinbase Historical
# ===================================================================
def ingest_coinbase(conn, pairs: list[str]) -> dict:
    """Bulk-load 1 year of OHLCV candles from Coinbase public API."""
    try:
        from coinbase_historical_feed import CoinbaseHistoricalFeed
    except ImportError as exc:
        return {"status": "skipped", "reason": f"import failed: {exc}"}

    _restore_stdout()
    print(f"\n[coinbase] Fetching historical candles for {pairs} ...")
    feed = CoinbaseHistoricalFeed()
    now = _now_ms()
    inserted = 0
    total_candles = 0

    try:
        all_data = feed.fetch_year_of_data(pairs)
    except Exception as exc:
        return {"status": "error", "reason": str(exc)}

    for pair, candles in all_data.items():
        pair_inserted = 0
        for c in candles:
            ts_ms = int(c.timestamp.timestamp() * 1000) if hasattr(c.timestamp, "timestamp") else int(c.timestamp)
            bar = {
                "provider": "coinbase",
                "venue": "coinbase",
                "symbol_id": f"coinbase:{pair}",
                "symbol": pair,
                "period_id": "1h",
                "time_start_ms": ts_ms,
                "time_end_ms": ts_ms + 3_600_000,  # 1 hour
                "open": c.open,
                "high": c.high,
                "low": c.low,
                "close": c.close,
                "volume": c.volume,
                "trades_count": None,
                "ingested_at_ms": now,
                "raw_json": None,
            }
            if db.insert_market_bar(conn, bar):
                pair_inserted += 1
        inserted += pair_inserted
        total_candles += len(candles)
        print(f"  [coinbase] {pair}: {pair_inserted} new / {len(candles)} total candles")
        conn.commit()

    db.set_watermark_ms(conn, "coinbase_historical", now)
    conn.commit()
    print(f"[coinbase] Inserted {inserted} market bars ({total_candles} candles processed).")
    return {"status": "ok", "inserted": inserted, "total_candles": total_candles}


# ===================================================================
# 6. Macro Intelligence
# ===================================================================
def ingest_macro_intel(conn) -> dict:
    """Store MacroIntelligence entry context as sentiment rows."""
    try:
        from macro_intelligence import MacroIntelligence
    except ImportError as exc:
        return {"status": "skipped", "reason": f"import failed: {exc}"}

    _restore_stdout()
    print("\n[macro_intel] Fetching macro intelligence context ...")
    mi = MacroIntelligence()
    now = _now_ms()
    inserted = 0

    # Query context for a few representative pairs to capture all sub-components
    probe_pairs = ["XBTUSD", "ETHUSD", "SOLUSD"]
    for pair in probe_pairs:
        try:
            ctx = mi.get_entry_context(pair)
        except Exception as exc:
            print(f"  [macro_intel] get_entry_context('{pair}') failed: {exc}")
            continue

        fields_to_store = [
            "fear_greed", "fear_label", "btc_24h", "eth_24h", "market_24h",
            "btc_dominance", "macro_score", "btc_vol_pattern", "is_trending",
        ]
        for field_name in fields_to_store:
            raw = ctx.get(field_name)
            if raw is None:
                continue
            numeric_value = None
            label = None
            if isinstance(raw, bool):
                numeric_value = 1.0 if raw else 0.0
                label = str(raw)
            elif isinstance(raw, (int, float)):
                numeric_value = float(raw)
            elif isinstance(raw, str):
                label = raw

            row = {
                "source": "macro_intelligence",
                "sentiment_id": _ts_id("macro_intel", now, f"{pair}:{field_name}"),
                "sentiment_type": field_name,
                "symbol": pair,
                "value": numeric_value,
                "label": label,
                "ts_ms": now,
                "ingested_at_ms": now,
                "raw_json": json.dumps({"pair": pair, "field": field_name, "raw": raw}),
            }
            if db.insert_sentiment(conn, row):
                inserted += 1

    conn.commit()
    db.set_watermark_ms(conn, "macro_intelligence", now)
    conn.commit()
    print(f"[macro_intel] Inserted {inserted} sentiment rows from macro intelligence.")
    return {"status": "ok", "inserted": inserted}


# ===================================================================
# CLI + orchestration
# ===================================================================
def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(
        description="Ingest existing Aureon feed modules into the global history SQLite DB.",
    )
    p.add_argument("--db", default=None, help="SQLite path override")
    p.add_argument(
        "--feeds", default="all",
        help="CSV of feeds to run: coingecko,macro,news,glassnode,coinbase,macro_intel,all  (default: all)",
    )
    p.add_argument(
        "--coinbase-pairs", default=",".join(DEFAULT_COINBASE_PAIRS),
        help="CSV of Coinbase pairs (default: %(default)s)",
    )
    p.add_argument(
        "--coingecko-coins", default=",".join(DEFAULT_COINGECKO_COINS),
        help="CSV of CoinGecko coin IDs",
    )
    p.add_argument(
        "--news-keywords", default=",".join(DEFAULT_NEWS_KEYWORDS),
        help="CSV of news search keywords (default: %(default)s)",
    )
    p.add_argument(
        "--hours-back", type=int, default=24,
        help="Hours of history for live feeds like Glassnode (default: 24)",
    )
    return p.parse_args()


def main() -> None:
    args = parse_args()

    feeds_requested = [f.strip().lower() for f in args.feeds.split(",")]
    if "all" in feeds_requested:
        feeds_requested = list(ALL_FEEDS)

    coins = [c.strip() for c in args.coingecko_coins.split(",") if c.strip()]
    pairs = [p.strip() for p in args.coinbase_pairs.split(",") if p.strip()]
    keywords = [k.strip() for k in args.news_keywords.split(",") if k.strip()]

    print("=" * 70)
    print("  Aureon Feed Ingestion -> Global History DB")
    print("=" * 70)
    print(f"  Feeds     : {', '.join(feeds_requested)}")
    print(f"  DB        : {args.db or '(default)'}")
    print(f"  Hours back: {args.hours_back}")
    print("=" * 70)

    conn = db.connect(args.db)
    results: dict[str, dict] = {}
    t0 = time.time()

    # Dispatch table
    dispatch = {
        "coingecko":  lambda: ingest_coingecko(conn, coins),
        "macro":      lambda: ingest_macro(conn),
        "news":       lambda: ingest_news(conn, keywords),
        "glassnode":  lambda: ingest_glassnode(conn, args.hours_back),
        "coinbase":   lambda: ingest_coinbase(conn, pairs),
        "macro_intel": lambda: ingest_macro_intel(conn),
    }

    for feed_name in feeds_requested:
        if feed_name not in dispatch:
            print(f"\n[{feed_name}] Unknown feed, skipping.")
            results[feed_name] = {"status": "skipped", "reason": "unknown feed name"}
            continue
        try:
            results[feed_name] = dispatch[feed_name]()
        except Exception as exc:
            print(f"\n[{feed_name}] FAILED: {exc}")
            results[feed_name] = {"status": "error", "reason": str(exc)}

    conn.close()
    elapsed = time.time() - t0

    # Summary
    print("\n" + "=" * 70)
    print("  INGESTION SUMMARY")
    print("=" * 70)
    for name, res in results.items():
        status = res.get("status", "?")
        detail = ""
        if status == "ok":
            parts = []
            for k, v in res.items():
                if k != "status":
                    parts.append(f"{k}={v}")
            detail = ", ".join(parts)
        elif status in ("skipped", "error"):
            detail = res.get("reason", "")
        print(f"  {name:15s}  {status:8s}  {detail}")
    print(f"\n  Total time: {elapsed:.1f}s")
    print("=" * 70)


if __name__ == "__main__":
    main()