Spaces:

GaryLeckey
/

aureon-os

Running

aureon-os / scripts /python /ingest_existing_feeds.py

Deploy Aureon OS (deploy/selfhost stack: nginx edge + fusion API x2 + console + HNC organ + redis + postgres; paper mode) (part 19)

da0ce7b verified 1 day ago

Raw History Blame Contribute Delete

23.6 kB

#!/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()

	#!/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()