Refine Argus market context and retention

.env.example

@@ -2,10 +2,11 @@
 # Copy this to .env and paste your real API keys.
 
 ARGUS_SQLITE_PATH=data/argus_mcp.sqlite3
-ARGUS_SYMBOLS=QQQ,SPY,DXY,HYG,BTCUSD,ETHUSD,VXX
+ARGUS_SYMBOLS=QQQ,SPY,XLK,SMH,HYG,VXX,UVXY,UUP,TLT,GLD,XLE,USO,IYT,JETS,ZIM,SBLK,DHT,CHRW,BTCUSD,ETHUSD
 ARGUS_INTERVAL=1d
 ARGUS_FINNHUB_TTL_SECONDS=60
-ARGUS_TWELVE_DATA_TTL_SECONDS=900
+ARGUS_TWELVE_DATA_TTL_SECONDS=10800
+ARGUS_SNAPSHOT_RETENTION_DAYS=30
 ARGUS_SNAPSHOT_TTL_SECONDS=60
 FINNHUB_TOKEN=
 TWELVE_DATA_KEY=

ARGUS_PAPER.md

@@ -0,0 +1,207 @@
+# Argus MCP paper
+
+## Purpose
+
+Argus MCP is a read-only market context service for the trader27 stack.
+
+It does not make trading decisions and it does not dispatch actions.
+Its job is to observe cross-market conditions and publish a compact, explainable description of the current environment so Hermes MCP can reason better.
+
+In short:
+- Argus senses
+- Hermes interprets and decides
+- Trader executes
+
+## What Argus provides
+
+Argus exposes two read-only products:
+- `get_snapshot()`
+- `get_regime()`
+
+### Snapshot
+
+The snapshot is the rich payload.
+It contains:
+- `snapshot_id`
+- `generated_at`
+- `regime`
+- `confidence`
+- `summary`
+- `components`
+- `signals`
+- `impacts`
+- `source_status`
+
+Use the snapshot when Hermes wants the full evidence trail.
+
+### Regime
+
+The regime payload is the compact payload.
+It contains:
+- `snapshot_id`
+- `generated_at`
+- `regime`
+- `confidence`
+- `summary`
+- `components`
+
+Use the regime payload when Hermes wants a small, decision-friendly context object.
+
+## Provider policy
+
+### Finnhub
+
+Finnhub is the primary live source.
+Use it for the fast layer and for most ETF, equity, volatility, transport, and crypto proxies.
+
+### Twelve Data
+
+Twelve Data is optional slow enrichment only.
+Use it only where Finnhub is missing something useful.
+It should be treated as sparse context, not the primary engine.
+
+Current policy:
+- long cache time, 3 hours
+- avoid frequent refreshes
+- prefer it only for selective fallback or enrichment symbols
+
+This means Argus should still remain useful even if Twelve Data is removed.
+
+## Current signal domains
+
+### 1. Risk appetite
+
+Purpose:
+measure speculative participation and growth leadership.
+
+Current proxies:
+- `QQQ`
+- `SPY`
+- `XLK`
+- `SMH`
+- `BTCUSD`
+- `ETHUSD`
+
+### 2. Stress
+
+Purpose:
+measure volatility and defensive positioning.
+
+Current proxies:
+- `VXX`
+- `UVXY`
+- `TLT`
+- inverse contribution from `HYG`
+
+### 3. Liquidity
+
+Purpose:
+measure whether conditions are supportive or tightening.
+
+Current proxies:
+- `UUP`
+- `HYG`
+- confirming support from `QQQ` and `BTCUSD`
+
+### 4. Real asset pressure
+
+Purpose:
+measure hard-asset, inflation, and commodity pressure.
+
+Current proxies:
+- `GLD`
+- `XLE`
+- `USO`
+
+### 5. Transport pressure
+
+Purpose:
+measure supply chain, freight, airline, logistics, and shipping stress.
+Transport weakness contributes to fear and stress because it can imply deteriorating demand, strained supply lines, or both.
+
+Current proxies:
+- `IYT`
+- `JETS`
+- `ZIM`
+- `SBLK`
+- `DHT`
+- `CHRW`
+
+## Reported states
+
+Argus currently emits these regime states:
+- `risk_on`
+- `risk_off`
+- `stress`
+- `compression`
+- `real_asset_inflation`
+- `neutral`
+- `no_data`
+
+### Meanings
+
+#### risk_on
+Speculative participation and liquidity are supportive.
+Useful for Hermes as a tailwind for breakout or active directional behavior.
+
+#### risk_off
+Defensive positioning is building.
+Useful for Hermes as a warning against aggressive participation.
+
+#### stress
+Volatility and transport or macro strain dominate.
+Useful for Hermes as a caution state with elevated fear and fragility.
+
+#### compression
+Signals are subdued, range-like, and not strongly directional.
+Useful for Hermes as a favorable background for grid or mean-reverting logic.
+
+#### real_asset_inflation
+Hard assets and commodities are leading relative to financial assets.
+Useful for Hermes as a sign that nominal risk may be mixed even if headline stress is not extreme.
+
+#### neutral
+Signals are mixed and the top regime is not clearly separated.
+Useful for Hermes as a low-conviction state.
+
+#### no_data
+Not enough inputs are available.
+Useful for Hermes as a degraded mode signal.
+
+## How Hermes should use Argus
+
+Hermes should use Argus as context, not as a command channel.
+
+Recommended pattern:
+1. read `regime`
+2. read `confidence`
+3. read `components`
+4. optionally inspect `snapshot` for evidence details
+
+### Suggested Hermes interpretation rules
+
+- `compression` + `range_bound` + `macro_stress_low`
+  - favor grid or range-harvesting logic
+- `risk_on` + `tech_leadership` + `liquidity_supportive`
+  - allow stronger directional bias, but still respect local trader evidence
+- `stress` + `transport_stress`
+  - reduce aggression, widen caution, avoid trusting short-term noise
+- `real_asset_inflation` + `gold_bid` + `energy_pressure_up`
+  - treat the market as late-cycle or inflation-sensitive, not cleanly bullish
+- `neutral` or `low_confidence`
+  - avoid overreacting to Argus alone
+
+## Design principles
+
+- keep the public surface small
+- keep the output explainable
+- prefer a few strong proxies over symbol sprawl
+- prefer cached, auditable data over rate-limit churn
+- let Hermes make decisions, not Argus
+
+## Next likely improvements
+
+- continue tuning transport versus stress weights
+- evaluate whether Twelve Data should remain enabled at all
+- later integrate metals-mcp or another dedicated hard-asset source for gold and related signals
+- stabilize the state enum contract before deeper Hermes integration

PROJECT.md

@@ -9,6 +9,7 @@ Provide Hermes with a small, explainable, read-only regime feed for market senti
 - FastAPI + FastMCP on one port
 - SQLite persistence for snapshots
 - provider adapters for Finnhub and Twelve Data
+- domain scoring for risk appetite, stress, liquidity, and real-asset pressure
 
 ## Scope guardrails
 - no trading decisions

README.md

@@ -1,11 +1,11 @@
 # Argus MCP
 
-Argus MCP is a narrow, read-only market context server for Hermes.
+Argus MCP is a read-only market context feed for Hermes. It watches cross-market proxies and publishes regime snapshots so Hermes can reason about market conditions without making decisions.
 
 ## What it does
 
 - ingests context from Finnhub and Twelve Data
-- classifies a market regime
+- classifies a market regime from domain scores (risk appetite, stress, liquidity, real-asset pressure, transport pressure)
 - stores immutable snapshots in SQLite
 - exposes read-only MCP tools
 

src/argus_mcp/config.py

@@ -4,6 +4,8 @@ from dataclasses import dataclass
 from pathlib import Path
 import os
 
+from argus_mcp.symbols import canonicalize_symbol
+
 
 def _load_dotenv(path: Path) -> None:
     if not path.exists():
@@ -54,25 +56,29 @@ class ArgusConfig:
     sqlite_path: Path = Path("data/argus_mcp.sqlite3")
     finnhub_token: str = ""
     twelve_data_key: str = ""
-    symbols: tuple[str, ...] = ("QQQ", "SPY", "DXY", "HYG", "BTCUSD", "ETHUSD", "VXX")
+    symbols: tuple[str, ...] = ("QQQ", "SPY", "XLK", "SMH", "HYG", "VXX", "UVXY", "UUP", "TLT", "GLD", "XLE", "USO", "IYT", "JETS", "ZIM", "SBLK", "DHT", "CHRW", "BTCUSD", "ETHUSD")
     interval: str = "1d"
     finnhub_ttl_seconds: int = 60
-    twelve_data_ttl_seconds: int = 900
+    twelve_data_ttl_seconds: int = 10800
     snapshot_ttl_seconds: int = 60
+    snapshot_retention_days: int = 30
 
 
 def load_config() -> ArgusConfig:
     _load_dotenv(Path.cwd() / ".env")
+    raw_symbols = _split_csv(
+        _env("ARGUS_SYMBOLS"),
+        ("QQQ", "SPY", "XLK", "SMH", "HYG", "VXX", "UVXY", "UUP", "TLT", "GLD", "XLE", "USO", "IYT", "JETS", "ZIM", "SBLK", "DHT", "CHRW", "BTCUSD", "ETHUSD"),
+    )
+    canonical_symbols = tuple(dict.fromkeys(canonicalize_symbol(symbol) for symbol in raw_symbols))
     return ArgusConfig(
         sqlite_path=Path(_env("ARGUS_SQLITE_PATH", "data/argus_mcp.sqlite3")),
         finnhub_token=_env("FINNHUB_TOKEN"),
         twelve_data_key=_env("TWELVE_DATA_KEY"),
-        symbols=_split_csv(
-            _env("ARGUS_SYMBOLS"),
-            ("QQQ", "SPY", "DXY", "HYG", "BTCUSD", "ETHUSD", "VXX"),
-        ),
+        symbols=canonical_symbols,
         interval=_env("ARGUS_INTERVAL", "1d"),
         finnhub_ttl_seconds=_int_env("ARGUS_FINNHUB_TTL_SECONDS", 60),
-        twelve_data_ttl_seconds=_int_env("ARGUS_TWELVE_DATA_TTL_SECONDS", 900),
+        twelve_data_ttl_seconds=_int_env("ARGUS_TWELVE_DATA_TTL_SECONDS", 10800),
         snapshot_ttl_seconds=_int_env("ARGUS_SNAPSHOT_TTL_SECONDS", 60),
+        snapshot_retention_days=_int_env("ARGUS_SNAPSHOT_RETENTION_DAYS", 30),
     )

src/argus_mcp/models.py

@@ -35,4 +35,3 @@ class RegimeSnapshot(BaseModel):
     signals: list[MarketQuote] = Field(default_factory=list)
     impacts: list[SignalImpact] = Field(default_factory=list)
     source_status: dict[str, str] = Field(default_factory=dict)
-

src/argus_mcp/regime.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from collections.abc import Iterable
 from datetime import datetime, timezone
+from math import tanh
 from uuid import uuid4
 
 from argus_mcp.models import MarketQuote, RegimeSnapshot, SignalImpact
@@ -13,7 +14,14 @@ def _change(quote: MarketQuote | None) -> float:
     return float(quote.change_pct)
 
 
-def _score_component(value: float, weight: float) -> float:
+def _norm(value: float, scale: float = 1.0) -> float:
+    if scale <= 0:
+        return 0.0
+    return tanh(value / scale)
+
+
+def _append_impact(impacts: list[SignalImpact], name: str, value: float, weight: float, note: str) -> float:
+    impacts.append(SignalImpact(name=name, value=value, weight=weight, note=note))
     return value * weight
 
 
@@ -23,82 +31,141 @@ def build_regime_snapshot(quotes: Iterable[MarketQuote]) -> RegimeSnapshot:
 
     qqq = by_symbol.get("QQQ")
     spy = by_symbol.get("SPY")
-    dxy = by_symbol.get("DXY") or by_symbol.get("UUP")
     hyg = by_symbol.get("HYG")
+    vxx = by_symbol.get("VXX")
+    uvxy = by_symbol.get("UVXY")
+    uup = by_symbol.get("UUP") or by_symbol.get("DXY")
+    tlt = by_symbol.get("TLT")
+    gld = by_symbol.get("GLD")
+    xle = by_symbol.get("XLE")
+    uso = by_symbol.get("USO")
+    xlk = by_symbol.get("XLK")
+    smh = by_symbol.get("SMH")
+    iyt = by_symbol.get("IYT")
+    jets = by_symbol.get("JETS")
+    zim = by_symbol.get("ZIM")
+    sblk = by_symbol.get("SBLK")
+    dht = by_symbol.get("DHT")
+    chrw = by_symbol.get("CHRW")
     btc = by_symbol.get("BTCUSD") or by_symbol.get("BTC/USD")
     eth = by_symbol.get("ETHUSD") or by_symbol.get("ETH/USD")
-    vxx = by_symbol.get("VXX")
 
-    risk = 0.0
-    stress = 0.0
-    liquidity = 0.0
-    compression = 0.0
     impacts: list[SignalImpact] = []
 
+    qqq_move = _norm(_change(qqq), 1.5)
+    spy_move = _norm(_change(spy), 1.5)
+    qqq_vs_spy = _norm((_change(qqq) - _change(spy)) * 2.0, 1.0)
+    btc_move = _norm(_change(btc), 2.5)
+    eth_vs_btc = _norm((_change(eth) - _change(btc)) * 2.0, 1.5)
+    hyg_move = _norm(_change(hyg), 0.8)
+    vxx_move = _norm(_change(vxx), 2.0)
+    uvxy_move = _norm(_change(uvxy), 3.0)
+    uup_move = _norm(_change(uup), 0.5)
+    tlt_move = _norm(_change(tlt), 0.8)
+    gld_move = _norm(_change(gld), 1.2)
+    xle_move = _norm(_change(xle), 1.8)
+
+    risk_appetite = 0.0
+    if qqq:
+        risk_appetite += _append_impact(impacts, "qqq_momentum", qqq_move, 0.22, "Growth equity momentum")
     if qqq and spy:
-        spread = _change(qqq) - _change(spy)
-        delta = _score_component(spread, 0.35)
-        risk += delta
-        impacts.append(SignalImpact(name="qqq_vs_spy", value=spread, weight=0.35, note="Speculative leadership spread"))
-
+        risk_appetite += _append_impact(impacts, "qqq_vs_spy", qqq_vs_spy, 0.28, "Speculative leadership over broad market")
     if btc:
-        delta = _score_component(_change(btc), 0.3)
-        risk += delta
-        impacts.append(SignalImpact(name="btc_momentum", value=_change(btc), weight=0.3, note="Crypto bid strength"))
-
-    if eth:
-        delta = _score_component(_change(eth), 0.25)
-        risk += delta
-        impacts.append(SignalImpact(name="eth_momentum", value=_change(eth), weight=0.25, note="Altcoin relative strength"))
+        risk_appetite += _append_impact(impacts, "btc_momentum", btc_move, 0.30, "Crypto beta appetite")
+    if btc and eth:
+        risk_appetite += _append_impact(impacts, "eth_vs_btc", eth_vs_btc, 0.20, "Altcoin breadth versus BTC")
+    if xlk:
+        risk_appetite += _append_impact(impacts, "xlk_momentum", _norm(_change(xlk), 1.2), 0.14, "Nasdaq / tech leadership")
+    if smh:
+        risk_appetite += _append_impact(impacts, "smh_momentum", _norm(_change(smh), 1.4), 0.16, "Semiconductor leadership")
 
-    if dxy:
-        delta = _score_component(_change(dxy), 0.45)
-        stress += delta
-        liquidity -= delta
-        impacts.append(SignalImpact(name="dollar_strength", value=_change(dxy), weight=0.45, note="Dollar pressure on liquidity"))
+    stress = 0.0
+    if vxx:
+        stress += _append_impact(impacts, "vxx_stress", vxx_move, 0.40, "Volatility demand proxy")
+    if uvxy:
+        stress += _append_impact(impacts, "uvxy_shock", uvxy_move, 0.25, "Acute volatility shock proxy")
+    if hyg:
+        stress += _append_impact(impacts, "hyg_inverse", -hyg_move, 0.20, "Credit weakness raises stress")
+    if tlt:
+        stress += _append_impact(impacts, "tlt_bid", tlt_move, 0.15, "Bond bid as defensive positioning")
+
+    transport_pressure = 0.0
+    if iyt:
+        transport_pressure += _append_impact(impacts, "iyt_inverse", -_norm(_change(iyt), 1.0), 0.22, "Transportation weakness")
+    if jets:
+        transport_pressure += _append_impact(impacts, "jets_inverse", -_norm(_change(jets), 1.0), 0.16, "Air travel / transport demand weakness")
+    if zim:
+        transport_pressure += _append_impact(impacts, "zim_inverse", -_norm(_change(zim), 2.0), 0.20, "Container shipping weakness")
+    if sblk:
+        transport_pressure += _append_impact(impacts, "sblk_inverse", -_norm(_change(sblk), 2.0), 0.14, "Dry bulk shipping weakness")
+    if dht:
+        transport_pressure += _append_impact(impacts, "dht_inverse", -_norm(_change(dht), 2.0), 0.14, "Tanker shipping weakness")
+    if chrw:
+        transport_pressure += _append_impact(impacts, "chrw_inverse", -_norm(_change(chrw), 1.5), 0.14, "Logistics / freight brokerage weakness")
 
+    liquidity = 0.0
+    if uup:
+        liquidity += _append_impact(impacts, "uup_inverse", -uup_move, 0.45, "Softer dollar supports liquidity")
     if hyg:
-        delta = _score_component(_change(hyg), 0.35)
-        liquidity += delta
-        stress -= delta
-        impacts.append(SignalImpact(name="credit_spread_proxy", value=_change(hyg), weight=0.35, note="Credit appetite proxy"))
+        liquidity += _append_impact(impacts, "hyg_support", hyg_move, 0.25, "Credit support for liquidity")
+    if qqq:
+        liquidity += _append_impact(impacts, "qqq_support", qqq_move, 0.15, "Equity strength confirms liquidity")
+    if btc:
+        liquidity += _append_impact(impacts, "btc_support", btc_move, 0.15, "Crypto participation confirms liquidity")
 
-    if vxx:
-        delta = _score_component(_change(vxx), 0.6)
-        stress += delta
-        compression -= abs(delta)
-        impacts.append(SignalImpact(name="volatility_proxy", value=_change(vxx), weight=0.6, note="Stress and vol demand proxy"))
+    real_asset_pressure = 0.0
+    if gld:
+        real_asset_pressure += _append_impact(impacts, "gld_strength", gld_move, 0.55, "Gold strength / hard-asset demand")
+    if xle:
+        real_asset_pressure += _append_impact(impacts, "xle_strength", xle_move, 0.45, "Energy / commodity cycle pressure")
+    if uso:
+        real_asset_pressure += _append_impact(impacts, "uso_strength", _norm(_change(uso), 3.0), 0.35, "Crude oil pressure / inflation impulse")
+
+    compression = max(0.0, 1.0 - (0.40 * abs(risk_appetite) + 0.30 * abs(stress) + 0.20 * abs(liquidity) + 0.10 * abs(transport_pressure)))
 
     if not quote_list:
         regime = "no_data"
         confidence = 0.0
         summary = "No provider data available yet."
+        regime_scores = {
+            "risk_on": 0.0,
+            "risk_off": 0.0,
+            "stress": 0.0,
+            "compression": 0.0,
+            "real_asset_inflation": 0.0,
+        }
     else:
-        scores = {
-            "risk_on": risk,
-            "stress": stress,
-            "liquidity": liquidity,
-            "compression": compression,
+        regime_scores = {
+            "risk_on": 0.50 * risk_appetite + 0.28 * liquidity - 0.25 * stress - 0.10 * real_asset_pressure - 0.12 * transport_pressure,
+            "risk_off": 0.40 * stress - 0.28 * risk_appetite - 0.18 * liquidity + 0.08 * real_asset_pressure + 0.26 * transport_pressure,
+            "stress": 0.50 * stress + 0.42 * transport_pressure - 0.15 * liquidity - 0.10 * risk_appetite,
+            "compression": compression - 0.12 * abs(real_asset_pressure) - 0.10 * abs(transport_pressure),
+            "real_asset_inflation": 0.60 * real_asset_pressure + 0.12 * stress + 0.08 * liquidity - 0.10 * risk_appetite,
         }
-        regime = max(scores, key=scores.get)
-        top_score = scores[regime]
-        if abs(top_score) < 0.25:
+        ranked = sorted(regime_scores.items(), key=lambda item: item[1], reverse=True)
+        regime, top_score = ranked[0]
+        second_score = ranked[1][1] if len(ranked) > 1 else 0.0
+        separation = top_score - second_score
+        if top_score < 0.18 or (separation < 0.08 and top_score < 0.55):
             regime = "neutral"
-            confidence = 0.1
-            summary = "Signals are mixed or too weak to call a regime with confidence."
+            confidence = max(0.1, min(0.35, top_score + max(separation, 0.0)))
+            summary = "Signals are mixed or weak, no regime has clear control."
         else:
-            confidence = min(1.0, max(0.1, abs(top_score) / 3.0))
+            confidence = max(0.15, min(0.95, 0.55 * top_score + 0.45 * separation))
             summary = {
-                "risk_on": "Speculative risk appetite is leading.",
-                "stress": "Volatility or funding stress is dominating.",
-                "liquidity": "Liquidity support is improving.",
-                "compression": "Market conditions look compressed and range-like.",
+                "risk_on": "Speculative risk appetite and liquidity are supportive for crypto.",
+                "risk_off": "Defensive positioning is building against crypto risk assets.",
+                "stress": "Volatility and macro stress are dominating the environment.",
+                "compression": "Cross-market signals are subdued and range-like.",
+                "real_asset_inflation": "Hard-asset and commodity pressure is rising relative to financial assets.",
             }[regime]
 
     components = {
-        "risk_on": risk,
+        "risk_appetite": risk_appetite,
         "stress": stress,
         "liquidity": liquidity,
+        "real_asset_pressure": real_asset_pressure,
+        "transport_pressure": transport_pressure,
         "compression": compression,
     }
 

src/argus_mcp/service.py

@@ -9,20 +9,7 @@ from argus_mcp.providers.finnhub import FinnhubProvider
 from argus_mcp.providers.twelve_data import TwelveDataProvider
 from argus_mcp.regime import build_regime_snapshot
 from argus_mcp.storage import SnapshotStore
-
-
-SYMBOL_ALIASES: dict[str, dict[str, str]] = {
-    "QQQ": {"finnhub": "QQQ", "twelve_data": "QQQ"},
-    "SPY": {"finnhub": "SPY", "twelve_data": "SPY"},
-    "HYG": {"finnhub": "HYG", "twelve_data": "HYG"},
-    "DXY": {"finnhub": "UUP", "twelve_data": "DXY"},
-    "UUP": {"finnhub": "UUP", "twelve_data": "UUP"},
-    "VXX": {"finnhub": "VXX", "twelve_data": "VXX"},
-    "BTCUSD": {"finnhub": "BINANCE:BTCUSDT", "twelve_data": "BTC/USD"},
-    "BTC/USD": {"finnhub": "BINANCE:BTCUSDT", "twelve_data": "BTC/USD"},
-    "ETHUSD": {"finnhub": "BINANCE:ETHUSDT", "twelve_data": "ETH/USD"},
-    "ETH/USD": {"finnhub": "BINANCE:ETHUSDT", "twelve_data": "ETH/USD"},
-}
+from argus_mcp.symbols import get_symbol_spec
 
 
 @dataclass(slots=True)
@@ -87,24 +74,32 @@ class ArgusService:
         quotes: list[MarketQuote] = []
         source_status: dict[str, str] = {}
         for symbol in self.config.symbols:
-            aliases = SYMBOL_ALIASES.get(symbol.upper(), {"finnhub": symbol, "twelve_data": symbol})
-            quote, status = await self._fetch_or_cache(
-                symbol,
-                self.finnhub,
-                aliases["finnhub"],
-                self.config.finnhub_ttl_seconds,
-            )
-            source_status[f"{symbol}:finnhub"] = status
-            if quote is None:
+            spec = get_symbol_spec(symbol)
+            quote = None
+
+            if spec.finnhub:
                 quote, status = await self._fetch_or_cache(
-                    symbol,
-                    self.twelve_data,
-                    aliases["twelve_data"],
-                    self.config.twelve_data_ttl_seconds,
+                    spec.canonical,
+                    self.finnhub,
+                    spec.finnhub,
+                    self.config.finnhub_ttl_seconds,
                 )
-                source_status[f"{symbol}:twelve_data"] = status
+                source_status[f"{spec.canonical}:finnhub"] = status
+
+            if quote is None:
+                if spec.twelve_data:
+                    quote, status = await self._fetch_or_cache(
+                        spec.canonical,
+                        self.twelve_data,
+                        spec.twelve_data,
+                        self.config.twelve_data_ttl_seconds,
+                    )
+                    source_status[f"{spec.canonical}:twelve_data"] = status
+                else:
+                    source_status[f"{spec.canonical}:twelve_data"] = "disabled:twelve_data"
+
             if quote is not None:
-                quote.symbol = symbol
+                quote.symbol = spec.canonical
                 quotes.append(quote)
         self._last_source_status = source_status
         return quotes
@@ -114,6 +109,7 @@ class ArgusService:
         snapshot = build_regime_snapshot(quotes)
         snapshot.source_status = dict(self._last_source_status)
         self.store.save(snapshot)
+        self.store.prune_snapshots_older_than(self.config.snapshot_retention_days)
         return snapshot
 
     async def get_snapshot(self, refresh: bool = False) -> RegimeSnapshot:

src/argus_mcp/storage.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 import json
 import sqlite3
 from pathlib import Path
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 
 from argus_mcp.models import MarketQuote
 from argus_mcp.models import RegimeSnapshot
@@ -65,6 +65,18 @@ class SnapshotStore:
             )
             conn.commit()
 
+    def prune_snapshots_older_than(self, days: int) -> int:
+        if days <= 0:
+            return 0
+        cutoff = datetime.now(timezone.utc) - timedelta(days=days)
+        with self._connect() as conn:
+            cursor = conn.execute(
+                "DELETE FROM snapshots WHERE generated_at < ?",
+                (cutoff.isoformat(),),
+            )
+            conn.commit()
+            return int(cursor.rowcount or 0)
+
     def latest(self) -> RegimeSnapshot | None:
         with self._connect() as conn:
             row = conn.execute(

src/argus_mcp/symbols.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, slots=True)
+class SymbolSpec:
+    canonical: str
+    finnhub: str | None = None
+    twelve_data: str | None = None
+    note: str = ""
+
+
+SYMBOL_REGISTRY: dict[str, SymbolSpec] = {
+    "QQQ": SymbolSpec("QQQ", finnhub="QQQ", twelve_data="QQQ", note="Equity growth / speculative appetite"),
+    "SPY": SymbolSpec("SPY", finnhub="SPY", twelve_data="SPY", note="Broad equity risk proxy"),
+    "HYG": SymbolSpec("HYG", finnhub="HYG", twelve_data="HYG", note="Credit / liquidity proxy"),
+    "VXX": SymbolSpec("VXX", finnhub="VXX", twelve_data="VXX", note="Volatility stress proxy"),
+    "UVXY": SymbolSpec("UVXY", finnhub="UVXY", twelve_data="UVXY", note="Acute volatility shock proxy"),
+    "UUP": SymbolSpec("UUP", finnhub="UUP", twelve_data="UUP", note="US dollar liquidity proxy"),
+    "DXY": SymbolSpec("UUP", finnhub="UUP", twelve_data=None, note="Legacy alias mapped to UUP"),
+    "TLT": SymbolSpec("TLT", finnhub="TLT", twelve_data="TLT", note="Long duration / risk-off proxy"),
+    "GLD": SymbolSpec("GLD", finnhub="GLD", twelve_data="GLD", note="Gold ETF proxy"),
+    "XLE": SymbolSpec("XLE", finnhub="XLE", twelve_data="XLE", note="Energy / commodity cycle proxy"),
+    "USO": SymbolSpec("USO", finnhub="USO", twelve_data="USO", note="Crude oil proxy"),
+    "XLK": SymbolSpec("XLK", finnhub="XLK", twelve_data="XLK", note="Nasdaq / tech sector proxy"),
+    "SMH": SymbolSpec("SMH", finnhub="SMH", twelve_data="SMH", note="Semiconductor leadership proxy"),
+    "IYT": SymbolSpec("IYT", finnhub="IYT", twelve_data="IYT", note="Transportation sector proxy"),
+    "JETS": SymbolSpec("JETS", finnhub="JETS", twelve_data="JETS", note="Air travel / transport demand proxy"),
+    "ZIM": SymbolSpec("ZIM", finnhub="ZIM", twelve_data="ZIM", note="Container shipping proxy"),
+    "SBLK": SymbolSpec("SBLK", finnhub="SBLK", twelve_data="SBLK", note="Dry bulk shipping proxy"),
+    "DHT": SymbolSpec("DHT", finnhub="DHT", twelve_data="DHT", note="Tanker shipping proxy"),
+    "CHRW": SymbolSpec("CHRW", finnhub="CHRW", twelve_data="CHRW", note="Logistics / freight brokerage proxy"),
+    "SLV": SymbolSpec("SLV", finnhub="SLV", twelve_data="SLV", note="Silver ETF proxy"),
+    "BTCUSD": SymbolSpec("BTCUSD", finnhub="BINANCE:BTCUSDT", twelve_data="BTC/USD", note="Crypto beta proxy"),
+    "BTC/USD": SymbolSpec("BTCUSD", finnhub="BINANCE:BTCUSDT", twelve_data="BTC/USD", note="Crypto beta proxy"),
+    "ETHUSD": SymbolSpec("ETHUSD", finnhub="BINANCE:ETHUSDT", twelve_data="ETH/USD", note="Crypto speculative breadth proxy"),
+    "ETH/USD": SymbolSpec("ETHUSD", finnhub="BINANCE:ETHUSDT", twelve_data="ETH/USD", note="Crypto speculative breadth proxy"),
+}
+
+
+def canonicalize_symbol(symbol: str) -> str:
+    key = symbol.strip().upper()
+    spec = SYMBOL_REGISTRY.get(key)
+    return spec.canonical if spec else key
+
+
+def get_symbol_spec(symbol: str) -> SymbolSpec:
+    key = symbol.strip().upper()
+    spec = SYMBOL_REGISTRY.get(key)
+    if spec is not None:
+        return spec
+    return SymbolSpec(canonical=key, finnhub=key, twelve_data=key)

tests/test_regime.py

@@ -7,12 +7,35 @@ def test_regime_prefers_risk_on_when_qqq_leads():
         [
             MarketQuote(symbol="QQQ", source="test", change_pct=2.0),
             MarketQuote(symbol="SPY", source="test", change_pct=0.5),
-            MarketQuote(symbol="BTCUSD", source="test", change_pct=1.0),
+            MarketQuote(symbol="BTCUSD", source="test", change_pct=2.5),
+            MarketQuote(symbol="ETHUSD", source="test", change_pct=4.0),
+            MarketQuote(symbol="HYG", source="test", change_pct=0.8),
+            MarketQuote(symbol="UUP", source="test", change_pct=-0.3),
+            MarketQuote(symbol="VXX", source="test", change_pct=-1.5),
         ]
     )
 
     assert snapshot.regime == "risk_on"
     assert snapshot.confidence > 0
+    assert snapshot.components["risk_appetite"] > 0
+
+
+def test_regime_prefers_stress_when_vol_spikes_and_credit_weakens():
+    snapshot = build_regime_snapshot(
+        [
+            MarketQuote(symbol="QQQ", source="test", change_pct=-1.5),
+            MarketQuote(symbol="SPY", source="test", change_pct=-0.9),
+            MarketQuote(symbol="BTCUSD", source="test", change_pct=-3.5),
+            MarketQuote(symbol="HYG", source="test", change_pct=-1.0),
+            MarketQuote(symbol="UUP", source="test", change_pct=0.7),
+            MarketQuote(symbol="VXX", source="test", change_pct=4.5),
+            MarketQuote(symbol="UVXY", source="test", change_pct=8.0),
+            MarketQuote(symbol="TLT", source="test", change_pct=1.0),
+        ]
+    )
+
+    assert snapshot.regime in {"stress", "risk_off"}
+    assert snapshot.components["stress"] > 0
 
 
 def test_regime_handles_no_data():

tests/test_storage.py

@@ -1,38 +1,34 @@
-from datetime import datetime, timezone
-from pathlib import Path
+from datetime import datetime, timezone, timedelta
 
 from argus_mcp.models import MarketQuote, RegimeSnapshot
 from argus_mcp.storage import SnapshotStore
 
 
-def test_storage_roundtrip(tmp_path: Path):
+def test_prune_snapshots_older_than(tmp_path):
     store = SnapshotStore(tmp_path / "argus.sqlite3")
-    snapshot = RegimeSnapshot(
-        snapshot_id="abc123",
+    old_snapshot = RegimeSnapshot(
+        snapshot_id="old",
+        generated_at=datetime.now(timezone.utc) - timedelta(days=31),
+        regime="neutral",
+        confidence=0.1,
+        summary="old",
+    )
+    fresh_snapshot = RegimeSnapshot(
+        snapshot_id="fresh",
         generated_at=datetime.now(timezone.utc),
-        regime="risk_on",
-        confidence=0.7,
-        summary="Test",
-        signals=[MarketQuote(symbol="QQQ", source="test", change_pct=1.0)],
+        regime="compression",
+        confidence=0.5,
+        summary="fresh",
+        signals=[MarketQuote(symbol="QQQ", source="finnhub")],
     )
 
-    store.save(snapshot)
+    store.save(old_snapshot)
+    store.save(fresh_snapshot)
+
+    deleted = store.prune_snapshots_older_than(30)
 
+    assert deleted == 1
+    assert store.count() == 1
     latest = store.latest()
     assert latest is not None
-    assert latest.snapshot_id == "abc123"
-    assert store.count() == 1
-
-
-def test_quote_cache_roundtrip(tmp_path: Path):
-    store = SnapshotStore(tmp_path / "argus.sqlite3")
-    quote = MarketQuote(symbol="QQQ", source="finnhub", last=500.0, change_pct=1.2)
-
-    store.save_quote(quote, fetched_at=datetime.now(timezone.utc))
-    cached = store.latest_quote("QQQ", "finnhub")
-
-    assert cached is not None
-    cached_quote, fetched_at = cached
-    assert cached_quote.symbol == "QQQ"
-    assert cached_quote.source == "finnhub"
-    assert fetched_at.tzinfo is not None
+    assert latest.snapshot_id == "fresh"

tests/test_symbols.py

@@ -0,0 +1,24 @@
+from argus_mcp.symbols import canonicalize_symbol, get_symbol_spec
+
+
+def test_canonicalize_legacy_dxy_to_uup():
+    assert canonicalize_symbol("DXY") == "UUP"
+
+
+def test_symbol_spec_maps_crypto_provider_symbols():
+    btc = get_symbol_spec("BTCUSD")
+    assert btc.canonical == "BTCUSD"
+    assert btc.finnhub == "BINANCE:BTCUSDT"
+    assert btc.twelve_data == "BTC/USD"
+
+
+def test_symbol_spec_covers_tech_and_energy_proxies():
+    assert get_symbol_spec("XLK").finnhub == "XLK"
+    assert get_symbol_spec("SMH").twelve_data == "SMH"
+    assert get_symbol_spec("USO").canonical == "USO"
+
+
+def test_symbol_spec_covers_transport_proxies():
+    assert get_symbol_spec("IYT").finnhub == "IYT"
+    assert get_symbol_spec("JETS").twelve_data == "JETS"
+    assert get_symbol_spec("ZIM").canonical == "ZIM"