Market Intelligence

Skill: Market Intelligence | Domain: trading | Category: research | Level: intermediate Tags: trading, research, macro, news, regime, intelligence

Market Intelligence — Complete Analysis Layer

The informational foundation of all trading decisions. Macro context → regime → news → sentiment → institutional → correlation → execute.

Sections

1. Macro Dashboard — DXY, VIX, yield curves, commodity-FX links, intermarket divergence
2. Regime Classifier — trending/ranging/volatile/quiet + strategy mapping
3. News & Events — economic calendar, impact scoring, event-price matching, sentiment
4. Sentiment & Positioning — retail positioning, COT, Fear/Greed, contrarian signals
5. Institutional Monitor — central banks, investment banks, COT analysis, intervention detection
6. Event Timeline — event linking, causal chains, narrative building, prediction
7. Pair Correlations — rolling correlation, divergence detection, clustering, regime switching
8. Trading Fundamentals — market structure, order types, asset classes, timeframes
9. Alternative Data — Google Trends signals, economic nowcasting, shipping/supply chain, search sentiment
10. Seasonality — monthly/day-of-week/hourly statistical edges with significance testing

Reference Files

• references/macro-regime.md — Macro dashboard (DXY/VIX/yield curves/commodities) + regime classifier (ADX/BB-width/MA-alignment + strategy map)
• references/news-sentiment.md — Economic calendar, news impact scoring, event-price matching, retail sentiment, COT, Fear/Greed, contrarian composite
• references/institutional-timeline.md — Central bank tracker, policy divergence, investment bank monitor, COT analyzer, intervention detector + full event timeline linker
• references/correlation-fundamentals.md — Pair correlation engine (rolling/historical/divergence/clustering/lead-lag) + correlation regime switcher + trading fundamentals reference
• references/alternative-data.md — AlternativeDataSources class: Google Trends signal, economic nowcast (Baltic Dry/copper-gold/credit spreads), search volume sentiment mapping
• references/seasonality.md — SeasonalityAnalyzer class: monthly/day-of-week/hourly return statistics with t-tests, p-values, win rates, and significance flags

Quick Decision Guide

Core Macro Intelligence Quick Card

RISK-ON:   VIX < 15, SPX rising → AUD, NZD up | JPY, CHF, Gold down
RISK-OFF:  VIX > 25, SPX falling → JPY, CHF, Gold up | AUD, NZD, EM down
DXY UP:    EUR, GBP, Gold down | USDJPY, USDCAD up
DXY DOWN:  EUR, GBP, Gold up | USDJPY, USDCAD down
YIELD ↑:   USD strengthens, Gold weakens
YIELD INVERSION (2-10 < 0): Recession warning, risk-off ahead
OIL ↑:     CAD, NOK strengthen
TREND REGIME:  ADX > 25, MAs aligned → use trend following
RANGE REGIME:  ADX < 20, BB narrow → use mean reversion
VOLATILE:  ADX < 20, BB wide → reduce size, wait for clarity

Implementations (Merged from sentiment-macro)

News Sentiment NLP Engine

import re, numpy as np
SENTIMENT_LEXICON = {
    "hawkish": 0.8, "dovish": -0.8, "rate hike": 0.7, "rate cut": -0.7,
    "inflation rises": 0.5, "inflation falls": -0.3, "recession": -0.8,
    "strong jobs": 0.6, "weak jobs": -0.6, "stimulus": 0.5, "tightening": 0.4,
    "crisis": -0.9, "default": -0.9, "war": -0.7, "peace": 0.3,
    "surge": 0.6, "plunge": -0.7, "rally": 0.5, "crash": -0.8,
    "beat expectations": 0.6, "miss expectations": -0.6, "surprise": 0.3,
    "upgrade": 0.5, "downgrade": -0.5, "bullish": 0.5, "bearish": -0.5,
}
class NewsSentimentNLP:
    @staticmethod
    def score_headline(headline: str) -> dict:
        h = headline.lower()
        matched = [(kw, score) for kw, score in SENTIMENT_LEXICON.items() if kw in h]
        avg = np.mean([s for _, s in matched]) if matched else 0
        entities = re.findall(r"\b(Fed|ECB|BOE|BOJ|NFP|CPI|GDP|FOMC|IMF)\b", headline, re.IGNORECASE)
        return {
            "headline": headline,
            "sentiment_score": round(avg, 3),
            "label": "BULLISH" if avg > 0.2 else "BEARISH" if avg < -0.2 else "NEUTRAL",
            "matched_keywords": [kw for kw, _ in matched],
            "entities": entities,
            "confidence": min(len(matched) / 3, 1.0),
        }
    @staticmethod
    def batch_score(headlines: list) -> dict:
        scores = [NewsSentimentNLP.score_headline(h) for h in headlines]
        avg = np.mean([s["sentiment_score"] for s in scores])
        return {"overall": round(avg, 3), "n_headlines": len(headlines),
                "bullish": sum(1 for s in scores if s["label"] == "BULLISH"),
                "bearish": sum(1 for s in scores if s["label"] == "BEARISH")}

Market News Impact

Overview

Monitors and analyzes major economic news, central bank decisions, geopolitical events, and market-moving developments. Matches news events to price reactions across instruments. Provides forward-looking event calendars with expected impact ratings.

Architecture

┌───────────────────────────────────────────────────────────┐
│                Market News Impact Engine                   │
├──────────────┬───────────────┬───────────────┬────────────┤
│ News Fetcher │ Event Calendar│ Impact Matcher│ Sentiment  │
│ & Classifier │ & Scheduler   │ & Scorer      │ Analyzer   │
└──────────────┴───────────────┴───────────────┴────────────┘

1. News Source Architecture

Source Priority (highest quality first)

NEWS_SOURCES = {
    "central_banks": {
        "fed":  {"url": "https://www.federalreserve.gov/newsevents.htm", "priority": 1},
        "ecb":  {"url": "https://www.ecb.europa.eu/press/html/index.en.html", "priority": 1},
        "boj":  {"url": "https://www.boj.or.jp/en/", "priority": 1},
        "boe":  {"url": "https://www.bankofengland.co.uk/news", "priority": 1},
        "rba":  {"url": "https://www.rba.gov.au/media-releases/", "priority": 1},
        "snb":  {"url": "https://www.snb.ch/en/", "priority": 1},
        "boc":  {"url": "https://www.bankofcanada.ca/press/", "priority": 1},
        "rbnz": {"url": "https://www.rbnz.govt.nz/news", "priority": 1},
    },
    "economic_data": {
        "forexfactory":  {"url": "https://www.forexfactory.com/calendar", "priority": 1},
        "investing_com":  {"url": "https://www.investing.com/economic-calendar/", "priority": 2},
        "tradingeconomics": {"url": "https://tradingeconomics.com/calendar", "priority": 2},
    },
    "financial_news": {
        "reuters":    {"url": "https://www.reuters.com/markets/", "priority": 1},
        "bloomberg":  {"url": "https://www.bloomberg.com/markets", "priority": 1},
        "wsj":        {"url": "https://www.wsj.com/news/markets", "priority": 2},
        "ft":         {"url": "https://www.ft.com/markets", "priority": 2},
        "cnbc":       {"url": "https://www.cnbc.com/world-markets/", "priority": 3},
    },
    "geopolitical": {
        "reuters_world": {"url": "https://www.reuters.com/world/", "priority": 1},
        "bbc_world":     {"url": "https://www.bbc.com/news/world", "priority": 2},
    },
}

News Fetching Framework

import requests
from bs4 import BeautifulSoup
from datetime import datetime, timedelta
from typing import Optional
import json

class NewsFetcher:
    """
    Fetch and classify market-moving news from multiple sources.
    In Claude context: use web_search tool as the primary fetcher.
    """

    # Impact classification keywords
    HIGH_IMPACT_KEYWORDS = [
        "rate decision", "interest rate", "nfp", "non-farm", "cpi", "inflation",
        "gdp", "fomc", "fed chair", "ecb president", "quantitative",
        "emergency", "war", "sanctions", "default", "recession", "crisis",
        "tariff", "trade war", "stimulus", "bailout"]
    MEDIUM_IMPACT_KEYWORDS = [
        "pmi", "employment", "retail sales", "housing", "trade balance",
        "industrial production", "consumer confidence", "jobless claims",
        "manufacturing", "services", "earnings", "ism"]

    @staticmethod
    def classify_impact(headline: str) -> str:
        """Classify a headline into impact level."""
        h = headline.lower()
        if any(kw in h for kw in NewsFetcher.HIGH_IMPACT_KEYWORDS):
            return "HIGH"
        if any(kw in h for kw in NewsFetcher.MEDIUM_IMPACT_KEYWORDS):
            return "MEDIUM"
        return "LOW"

    @staticmethod
    def extract_affected_currencies(headline: str) -> list[str]:
        """Extract which currencies are likely affected by a headline."""
        currency_map = {
            "fed": ["USD"], "fomc": ["USD"], "nfp": ["USD"], "us ": ["USD"],
            "ecb": ["EUR"], "euro": ["EUR"], "eurozone": ["EUR"],
            "boe": ["GBP"], "uk ": ["GBP"], "britain": ["GBP"], "sterling": ["GBP"],
            "boj": ["JPY"], "japan": ["JPY"], "yen": ["JPY"],
            "rba": ["AUD"], "australia": ["AUD"],
            "boc": ["CAD"], "canada": ["CAD"],
            "snb": ["CHF"], "swiss": ["CHF"],
            "rbnz": ["NZD"], "zealand": ["NZD"],
            "china": ["CNH", "AUD", "NZD"], "oil": ["CAD", "NOK"],
            "gold": ["XAU", "AUD"], "bitcoin": ["BTC"], "crypto": ["BTC", "ETH"],
        }
        h = headline.lower()
        affected = set()
        for trigger, currencies in currency_map.items():
            if trigger in h:
                affected.update(currencies)
        return list(affected) if affected else ["BROAD"]

    @staticmethod
    def map_to_pairs(currencies: list[str]) -> list[str]:
        """Map affected currencies to specific tradeable pairs."""
        major_pairs = {
            "USD": ["EURUSD", "GBPUSD", "USDJPY", "USDCHF", "AUDUSD", "USDCAD", "NZDUSD"],
            "EUR": ["EURUSD", "EURJPY", "EURGBP", "EURAUD", "EURCHF"],
            "GBP": ["GBPUSD", "EURGBP", "GBPJPY", "GBPAUD"],
            "JPY": ["USDJPY", "EURJPY", "GBPJPY", "AUDJPY"],
            "AUD": ["AUDUSD", "EURAUD", "GBPAUD", "AUDJPY", "AUDNZD"],
            "CAD": ["USDCAD", "CADCHF", "CADJPY"],
            "CHF": ["USDCHF", "EURCHF", "GBPCHF"],
            "NZD": ["NZDUSD", "AUDNZD", "NZDJPY"],
            "XAU": ["XAUUSD"],
            "BTC": ["BTCUSD"],
        }
        pairs = set()
        for c in currencies:
            pairs.update(major_pairs.get(c, []))
        return list(pairs)

2. Economic Calendar Engine

# Standard economic calendar event structure
EVENT_IMPACT_MAP = {
    # USD events
    "Non-Farm Payrolls":         {"impact": "HIGH", "currencies": ["USD"], "volatility_multiplier": 3.0},
    "FOMC Rate Decision":        {"impact": "HIGH", "currencies": ["USD"], "volatility_multiplier": 4.0},
    "FOMC Press Conference":     {"impact": "HIGH", "currencies": ["USD"], "volatility_multiplier": 3.5},
    "CPI m/m":                   {"impact": "HIGH", "currencies": ["USD"], "volatility_multiplier": 2.5},
    "CPI y/y":                   {"impact": "HIGH", "currencies": ["USD"], "volatility_multiplier": 2.5},
    "Core CPI":                  {"impact": "HIGH", "currencies": ["USD"], "volatility_multiplier": 2.5},
    "GDP q/q":                   {"impact": "HIGH", "currencies": ["USD"], "volatility_multiplier": 2.0},
    "Unemployment Rate":         {"impact": "HIGH", "currencies": ["USD"], "volatility_multiplier": 2.0},
    "ISM Manufacturing PMI":     {"impact": "MEDIUM", "currencies": ["USD"], "volatility_multiplier": 1.5},
    "ISM Services PMI":          {"impact": "MEDIUM", "currencies": ["USD"], "volatility_multiplier": 1.5},
    "Retail Sales":              {"impact": "MEDIUM", "currencies": ["USD"], "volatility_multiplier": 1.5},
    "Initial Jobless Claims":    {"impact": "MEDIUM", "currencies": ["USD"], "volatility_multiplier": 1.2},
    "Fed Chair Speech":          {"impact": "HIGH", "currencies": ["USD"], "volatility_multiplier": 3.0},
    # EUR events
    "ECB Rate Decision":         {"impact": "HIGH", "currencies": ["EUR"], "volatility_multiplier": 4.0},
    "ECB Press Conference":      {"impact": "HIGH", "currencies": ["EUR"], "volatility_multiplier": 3.5},
    "Eurozone CPI":              {"impact": "HIGH", "currencies": ["EUR"], "volatility_multiplier": 2.0},
    "German Manufacturing PMI":  {"impact": "MEDIUM", "currencies": ["EUR"], "volatility_multiplier": 1.3},
    # GBP events
    "BOE Rate Decision":         {"impact": "HIGH", "currencies": ["GBP"], "volatility_multiplier": 4.0},
    "UK CPI":                    {"impact": "HIGH", "currencies": ["GBP"], "volatility_multiplier": 2.0},
    # JPY events
    "BOJ Rate Decision":         {"impact": "HIGH", "currencies": ["JPY"], "volatility_multiplier": 4.0},
    "Japan CPI":                 {"impact": "MEDIUM", "currencies": ["JPY"], "volatility_multiplier": 1.5},
}

class EconomicCalendar:
    """Structured economic calendar with impact ratings and pair mapping."""

    def __init__(self):
        self.events = []

    def add_event(self, name: str, dt: datetime, actual: Optional[float] = None,
                  forecast: Optional[float] = None, previous: Optional[float] = None) -> dict:
        meta = EVENT_IMPACT_MAP.get(name, {"impact": "LOW", "currencies": ["BROAD"], "volatility_multiplier": 1.0})
        event = {
            "name": name,
            "datetime": dt.isoformat(),
            "actual": actual,
            "forecast": forecast,
            "previous": previous,
            "surprise": (actual - forecast) if actual is not None and forecast is not None else None,
            "impact": meta["impact"],
            "currencies": meta["currencies"],
            "affected_pairs": NewsFetcher.map_to_pairs(meta["currencies"]),
            "volatility_multiplier": meta["volatility_multiplier"],
        }
        self.events.append(event)
        return event

    def get_upcoming(self, hours: int = 24, impact_filter: Optional[str] = None) -> list[dict]:
        """Get events in the next N hours, optionally filtered by impact."""
        cutoff = datetime.utcnow() + timedelta(hours=hours)
        upcoming = [e for e in self.events if datetime.fromisoformat(e["datetime"]) <= cutoff]
        if impact_filter:
            upcoming = [e for e in upcoming if e["impact"] == impact_filter.upper()]
        return sorted(upcoming, key=lambda x: x["datetime"])

    def get_events_for_pair(self, pair: str) -> list[dict]:
        """All events affecting a specific pair."""
        return [e for e in self.events if pair.upper() in e["affected_pairs"]]

3. Price Impact Matching

def measure_event_impact(
    df: pd.DataFrame,
    event_time: datetime,
    pre_window_bars: int = 5,
    post_window_bars: int = 20,
) -> dict:
    """
    Measure price reaction around a specific event.
    Returns: pre-event range, immediate reaction, sustained move, retracement.
    """
    # Find nearest bar to event time
    idx = df.index.searchsorted(event_time)
    if idx < pre_window_bars or idx + post_window_bars > len(df):
        return {"error": "Insufficient data around event"}

    pre = df.iloc[idx - pre_window_bars:idx]
    post = df.iloc[idx:idx + post_window_bars]
    event_bar = df.iloc[idx]

    pre_range = pre["high"].max() - pre["low"].min()
    immediate_move = event_bar["close"] - pre.iloc[-1]["close"]
    max_post_move = post["high"].max() - pre.iloc[-1]["close"]
    min_post_move = post["low"].min() - pre.iloc[-1]["close"]
    sustained_move = post.iloc[-1]["close"] - pre.iloc[-1]["close"]

    return {
        "event_time": event_time.isoformat(),
        "pre_event_range": round(pre_range, 5),
        "immediate_move_pips": round(immediate_move / df.attrs.get("point", 0.0001), 1),
        "max_favorable": round(max_post_move / df.attrs.get("point", 0.0001), 1),
        "max_adverse": round(min_post_move / df.attrs.get("point", 0.0001), 1),
        "sustained_move_pips": round(sustained_move / df.attrs.get("point", 0.0001), 1),
        "retracement_pct": round(1 - abs(sustained_move) / max(abs(max_post_move), abs(min_post_move), 1e-10), 3),
        "direction": "bullish" if sustained_move > 0 else "bearish",
        "volatility_expansion": round(post["high"].max() - post["low"].min(), 5) / max(pre_range, 1e-10),
    }

def historical_event_impact_study(
    df: pd.DataFrame,
    events: list[dict],
    post_window: int = 20,
) -> pd.DataFrame:
    """
    Analyze price reaction across multiple historical occurrences of an event.
    Builds a statistical profile of how a pair reacts to a specific event type.
    """
    results = []
    for event in events:
        try:
            impact = measure_event_impact(df, datetime.fromisoformat(event["datetime"]), post_window_bars=post_window)
            impact["event_name"] = event["name"]
            impact["surprise"] = event.get("surprise")
            results.append(impact)
        except Exception:
            continue
    return pd.DataFrame(results)

4. Sentiment Analysis Framework

SENTIMENT_KEYWORDS = {
    "hawkish": +1.0, "tightening": +0.8, "rate hike": +0.9, "inflation concerns": +0.5,
    "strong employment": +0.5, "above expectations": +0.6,
    "dovish": -1.0, "easing": -0.8, "rate cut": -0.9, "slowdown": -0.5,
    "recession fears": -0.7, "below expectations": -0.6, "miss": -0.5,
    "risk on": +0.3, "rally": +0.4, "bullish": +0.5, "upgrade": +0.4,
    "risk off": -0.3, "sell-off": -0.5, "bearish": -0.5, "downgrade": -0.4,
    "uncertainty": -0.2, "volatile": -0.1, "crisis": -0.8, "default": -0.9,
    "stimulus": +0.6, "infrastructure": +0.3, "trade deal": +0.4,
    "sanctions": -0.4, "tariff": -0.3, "war": -0.8,
}

def score_sentiment(text: str) -> dict:
    """Quick keyword-based sentiment score for a news headline or summary."""
    text_lower = text.lower()
    scores = []
    matched = []
    for keyword, score in SENTIMENT_KEYWORDS.items():
        if keyword in text_lower:
            scores.append(score)
            matched.append(keyword)
    avg = np.mean(scores) if scores else 0.0
    return {
        "sentiment_score": round(avg, 3),
        "label": "BULLISH" if avg > 0.2 else "BEARISH" if avg < -0.2 else "NEUTRAL",
        "matched_keywords": matched,
        "confidence": min(len(matched) / 3, 1.0),
    }

def aggregate_sentiment(headlines: list[str]) -> dict:
    """Aggregate sentiment across multiple headlines."""
    scores = [score_sentiment(h) for h in headlines]
    avg_score = np.mean([s["sentiment_score"] for s in scores])
    return {
        "overall_score": round(avg_score, 3),
        "overall_label": "BULLISH" if avg_score > 0.15 else "BEARISH" if avg_score < -0.15 else "NEUTRAL",
        "n_headlines": len(headlines),
        "bullish_count": sum(1 for s in scores if s["label"] == "BULLISH"),
        "bearish_count": sum(1 for s in scores if s["label"] == "BEARISH"),
        "neutral_count": sum(1 for s in scores if s["label"] == "NEUTRAL"),
    }

5. Usage with Claude's Web Search

When this skill is active in Claude, the primary data acquisition method is web_search:

# Fetch current market news
web_search("forex market news today major events")
web_search("FOMC rate decision 2025")
web_search("forex factory economic calendar this week")
web_search("Reuters forex market analysis")

# Fetch specific event impact
web_search("NFP non-farm payrolls result today")
web_search("ECB interest rate decision impact EURUSD")

# Geopolitical
web_search("geopolitical risk forex markets today")
web_search("US China trade tensions impact currencies")

Then pipe fetched content through classify_impact(), extract_affected_currencies(), and score_sentiment() to produce structured, actionable output.

Integration Points

News Straddle Strategy

import pandas as pd, numpy as np

class NewsStraddleStrategy:

    @staticmethod
    def pre_news_straddle(current_price: float, atr: float, spread_pips: float) -> dict:
        """Place pending orders both sides before high-impact news."""
        buffer = atr * 0.5
        return {
            "strategy": "pre_news_straddle",
            "buy_stop": round(current_price + buffer, 5),
            "sell_stop": round(current_price - buffer, 5),
            "buy_sl": round(current_price, 5),
            "sell_sl": round(current_price, 5),
            "buy_tp": round(current_price + buffer + atr * 2, 5),
            "sell_tp": round(current_price - buffer - atr * 2, 5),
            "timing": "Place 2-5 minutes before news release",
            "cancel_unfilled": "Remove unfilled order immediately after news hits",
            "WARNING": "Spread widens massively during news. Slippage is real. Use limit orders where possible.",
            "risk": "HIGH — only use with 0.5% risk max",
        }

    @staticmethod
    def spike_fade(spike_direction: str, spike_high: float, spike_low: float, atr: float) -> dict:
        """Fade the initial news spike after it overextends."""
        if spike_direction == "up":
            entry = round(spike_high - atr * 0.3, 5)
            sl = round(spike_high + atr * 0.5, 5)
            tp = round(spike_high - atr * 1.5, 5)
        else:
            entry = round(spike_low + atr * 0.3, 5)
            sl = round(spike_low - atr * 0.5, 5)
            tp = round(spike_low + atr * 1.5, 5)
        return {
            "strategy": "spike_fade",
            "entry": entry, "sl": sl, "tp": tp,
            "direction": "SELL" if spike_direction == "up" else "BUY",
            "timing": "Wait 5-15 minutes after spike for momentum to exhaust",
            "confirmation": "Look for rejection candle (pin bar, engulfing) at spike extreme",
            "win_rate": "~55-60% historically — initial spikes retrace 50-70% of the move",
        }

    @staticmethod
    def news_momentum(data_surprise: float, direction: str, atr: float, entry_price: float) -> dict:
        """Ride the momentum when data significantly beats/misses expectations."""
        if abs(data_surprise) < 0.5:
            return {"signal": "NO TRADE — data in line with expectations, no directional edge"}
        strength = "STRONG" if abs(data_surprise) > 2 else "MODERATE"
        return {
            "strategy": "news_momentum",
            "surprise_magnitude": round(data_surprise, 2),
            "direction": direction,
            "strength": strength,
            "entry": round(entry_price, 5),
            "sl": round(entry_price - atr * 1.5, 5) if direction == "BUY" else round(entry_price + atr * 1.5, 5),
            "tp": round(entry_price + atr * 3, 5) if direction == "BUY" else round(entry_price - atr * 3, 5),
            "hold": "30 min to 4 hours depending on follow-through",
        }

Sentiment Extreme Contrarian

class SentimentContrarian:
    @staticmethod
    def composite_contrarian_signal(retail_long_pct: float, fear_greed: int, cot_percentile: float,
                                      social_bullish_pct: float) -> dict:
        scores = []
        if retail_long_pct > 75: scores.append(-0.8)
        elif retail_long_pct < 25: scores.append(0.8)
        else: scores.append(0)

        if fear_greed > 80: scores.append(-0.6)
        elif fear_greed < 20: scores.append(0.6)
        else: scores.append(0)

        if cot_percentile > 90: scores.append(-0.7)
        elif cot_percentile < 10: scores.append(0.7)
        else: scores.append(0)

        if social_bullish_pct > 80: scores.append(-0.5)
        elif social_bullish_pct < 20: scores.append(0.5)
        else: scores.append(0)

        import numpy as np
        avg = np.mean(scores)
        extreme_count = sum(1 for s in scores if abs(s) > 0.4)
        return {
            "composite_score": round(avg, 3),
            "direction": "CONTRARIAN BUY" if avg > 0.3 else "CONTRARIAN SELL" if avg < -0.3 else "NO EXTREME",
            "extreme_indicators": extreme_count,
            "conviction": "HIGH" if extreme_count >= 3 else "MODERATE" if extreme_count >= 2 else "LOW",
            "note": "Best when 3+ indicators at extremes simultaneously. Single indicator extremes are noisy.",
        }

Economic Indicator Tracker

INDICATORS = {
    "leading": [
        {"name": "PMI Manufacturing", "frequency": "monthly", "impact": "HIGH", "pairs": ["USD", "EUR", "GBP"]},
        {"name": "Building Permits", "frequency": "monthly", "impact": "MEDIUM", "pairs": ["USD"]},
        {"name": "Consumer Confidence", "frequency": "monthly", "impact": "MEDIUM", "pairs": ["USD", "EUR"]},
        {"name": "Yield Curve 2-10", "frequency": "daily", "impact": "HIGH", "pairs": ["USD"]},
        {"name": "New Orders Index", "frequency": "monthly", "impact": "MEDIUM", "pairs": ["USD"]},
        {"name": "Stock Market (SPX)", "frequency": "daily", "impact": "HIGH", "pairs": ["ALL"]},
        {"name": "Initial Jobless Claims", "frequency": "weekly", "impact": "MEDIUM", "pairs": ["USD"]}],
    "coincident": [
        {"name": "Non-Farm Payrolls", "frequency": "monthly", "impact": "HIGH", "pairs": ["USD"]},
        {"name": "Industrial Production", "frequency": "monthly", "impact": "MEDIUM", "pairs": ["USD", "EUR"]},
        {"name": "Retail Sales", "frequency": "monthly", "impact": "HIGH", "pairs": ["USD", "GBP"]},
        {"name": "GDP", "frequency": "quarterly", "impact": "HIGH", "pairs": ["ALL"]}],
    "lagging": [
        {"name": "CPI / Inflation", "frequency": "monthly", "impact": "HIGH", "pairs": ["ALL"]},
        {"name": "Unemployment Rate", "frequency": "monthly", "impact": "HIGH", "pairs": ["USD"]},
        {"name": "Core PCE", "frequency": "monthly", "impact": "HIGH", "pairs": ["USD"]},
        {"name": "Average Hourly Earnings", "frequency": "monthly", "impact": "MEDIUM", "pairs": ["USD"]}],
}

class EconomicIndicatorTracker:
    @staticmethod
    def cycle_position(leading_trend: str, coincident_trend: str, lagging_trend: str) -> dict:
        if leading_trend == "improving" and coincident_trend == "improving":
            phase = "EXPANSION — risk-on currencies favored (AUD, NZD, CAD)"
        elif leading_trend == "deteriorating" and coincident_trend == "improving":
            phase = "LATE CYCLE — be cautious, peak may be near"
        elif leading_trend == "deteriorating" and coincident_trend == "deteriorating":
            phase = "CONTRACTION — safe havens favored (JPY, CHF, USD, Gold)"
        elif leading_trend == "improving" and coincident_trend == "deteriorating":
            phase = "EARLY RECOVERY — selective risk-on, high-beta currencies"
        else:
            phase = "TRANSITION — mixed signals"
        return {"phase": phase, "leading": leading_trend, "coincident": coincident_trend, "lagging": lagging_trend}

    @staticmethod
    def surprise_index(actual: float, forecast: float, previous: float) -> dict:
        surprise = actual - forecast
        beat = actual > forecast
        return {
            "surprise": round(surprise, 3),
            "beat_expectations": beat,
            "vs_previous": "improving" if actual > previous else "deteriorating",
            "market_impact": "Positive surprise — currency should strengthen" if beat else "Negative surprise — currency weakens",
        }

COT Positioning Strategy

import pandas as pd, numpy as np

class COTStrategy:
    @staticmethod
    def weekly_change_signal(net_spec: int, prev_net_spec: int, percentile: float) -> dict:
        change = net_spec - prev_net_spec
        return {
            "strategy": "cot_positioning",
            "net_speculative": net_spec, "weekly_change": change,
            "percentile": round(percentile, 1),
            "signal": "CONTRARIAN SELL" if percentile > 90 and change < 0 else
                     "CONTRARIAN BUY" if percentile < 10 and change > 0 else
                     "TREND BUY" if change > 0 and 30 < percentile < 70 else
                     "TREND SELL" if change < 0 and 30 < percentile < 70 else "WAIT",
            "logic": "Extremes = fade the crowd. Mid-range = follow the smart money flow.",
            "timing": "Hold 2-8 weeks. COT is a slow signal — not for day trading.",
        }

Seasonality Analyzer

import pandas as pd
import numpy as np
from scipy import stats

class SeasonalityAnalyzer:

    @staticmethod
    def monthly_seasonality(df: pd.DataFrame) -> pd.DataFrame:
        """Monthly return statistics with significance testing."""
        df = df.copy()
        df["return"] = df["close"].pct_change()
        df["month"] = df.index.month
        monthly = df.groupby("month")["return"].agg(["mean", "std", "count"])
        monthly["annualized"] = monthly["mean"] * 21 * 12 * 100
        monthly["t_stat"] = monthly["mean"] / (monthly["std"] / np.sqrt(monthly["count"]))
        monthly["p_value"] = monthly["t_stat"].apply(lambda t: 2 * (1 - stats.t.cdf(abs(t), df=max(monthly["count"].min()-1, 1))))
        monthly["significant"] = monthly["p_value"] < 0.05
        monthly["win_rate"] = df.groupby("month")["return"].apply(lambda x: (x > 0).mean()) * 100
        monthly.index = ["Jan","Feb","Mar","Apr","May","Jun","Jul","Aug","Sep","Oct","Nov","Dec"]
        return monthly.round(4)

    @staticmethod
    def day_of_week_seasonality(df: pd.DataFrame) -> pd.DataFrame:
        df = df.copy()
        df["return"] = df["close"].pct_change()
        df["dow"] = df.index.dayofweek
        daily = df.groupby("dow")["return"].agg(["mean", "std", "count"])
        daily["annualized"] = daily["mean"] * 252 * 100
        daily["t_stat"] = daily["mean"] / (daily["std"] / np.sqrt(daily["count"]))
        daily["p_value"] = daily["t_stat"].apply(lambda t: 2 * (1 - stats.t.cdf(abs(t), df=max(daily["count"].min()-1, 1))))
        daily["significant"] = daily["p_value"] < 0.05
        daily.index = ["Mon", "Tue", "Wed", "Thu", "Fri"]
        return daily.round(4)

    @staticmethod
    def hourly_seasonality(df: pd.DataFrame) -> pd.DataFrame:
        df = df.copy()
        df["return"] = df["close"].pct_change()
        df["hour"] = df.index.hour
        hourly = df.groupby("hour")["return"].agg(["mean", "std", "count"])
        hourly["annualized"] = hourly["mean"] * 252 * 24 * 100
        hourly["range_pips"] = df.groupby(df.index.hour).apply(lambda x: (x["high"] - x["low"]).mean()) * 10000
        return hourly.round(4)

    @staticmethod
    def full_report(df: pd.DataFrame, symbol: str = "") -> dict:
        return {
            "symbol": symbol,
            "monthly": SeasonalityAnalyzer.monthly_seasonality(df).to_dict(),
            "daily": SeasonalityAnalyzer.day_of_week_seasonality(df).to_dict(),
            "hourly": SeasonalityAnalyzer.hourly_seasonality(df).to_dict(),
            "WARNING": "Seasonality = historical tendency, not guarantee. Always combine with other analysis.",
        }

Event Timeline Linker

Overview

Links data points across time from multiple sources (price, news, institutional flows, economic data, correlation shifts) to reconstruct event chains and build predictive narratives. Answers: What happened? Why? How? What was the result? What will likely happen next?

Architecture

┌─────────────────────────────────────────────────────────────────────┐
│                    Event Timeline Linker                            │
├────────────┬────────────┬──────────────┬───────────┬───────────────┤
│ Event      │ Temporal   │ Causal Chain │ Pattern   │ Prediction    │
│ Collector  │ Aligner    │ Builder      │ Matcher   │ Engine        │
└────────────┴────────────┴──────────────┴───────────┴───────────────┘
         ↑            ↑             ↑            ↑
    [mt5-chart]  [news-impact]  [correlation]  [institutional]

1. Event Data Model

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Optional, Literal
from dataclasses import dataclass, field, asdict
import json

@dataclass
class TimelineEvent:
    """Universal event structure for cross-source linking."""
    timestamp: datetime
    source: Literal["price", "news", "economic", "institutional", "correlation", "technical", "geopolitical"]
    event_type: str          # e.g., "rate_decision", "breakout", "correlation_shift"
    title: str               # human-readable title
    description: str         # detailed description
    impact: Literal["HIGH", "MEDIUM", "LOW"]
    affected_instruments: list[str] = field(default_factory=list)
    data: dict = field(default_factory=dict)  # source-specific payload
    tags: list[str] = field(default_factory=list)

    def to_dict(self) -> dict:
        d = asdict(self)
        d["timestamp"] = self.timestamp.isoformat()
        return d

@dataclass
class CausalLink:
    """Represents a cause -> effect relationship between events."""
    cause: TimelineEvent
    effect: TimelineEvent
    confidence: float        # 0-1 confidence in the causal relationship
    lag_seconds: int         # time between cause and effect
    mechanism: str           # how the cause led to the effect
    evidence: list[str] = field(default_factory=list)

2. Event Collector — Gather from All Sources

class EventCollector:
    """
    Collect and normalize events from all trading skill sources into
    a unified timeline. Each source adapter converts raw data into TimelineEvent.
    """

    @staticmethod
    def from_price_data(df: pd.DataFrame, symbol: str) -> list[TimelineEvent]:
        """Extract significant price events: breakouts, reversals, gaps, spikes."""
        events = []
        returns = df["close"].pct_change()
        atr_val = ((df["high"] - df["low"]).rolling(14).mean())

        for i in range(20, len(df)):
            row = df.iloc[i]
            ret = returns.iloc[i]

            # Large move detection (>2 ATR)
            move = abs(row["close"] - row["open"])
            if move > 2 * atr_val.iloc[i]:
                events.append(TimelineEvent(
                    timestamp=df.index[i],
                    source="price",
                    event_type="large_move",
                    title=f"{symbol} {'bullish' if ret > 0 else 'bearish'} spike",
                    description=f"{symbol} moved {abs(ret)*100:.2f}% in one bar ({move:.5f} > 2x ATR)",
                    impact="HIGH" if abs(ret) > 0.01 else "MEDIUM",
                    affected_instruments=[symbol],
                    data={"return": round(ret, 6), "atr": round(atr_val.iloc[i], 6)},
                    tags=["spike", "volatility"],
                ))

            # Gap detection
            if i > 0:
                gap = abs(row["open"] - df.iloc[i - 1]["close"])
                if gap > 1.5 * atr_val.iloc[i]:
                    events.append(TimelineEvent(
                        timestamp=df.index[i],
                        source="price",
                        event_type="gap",
                        title=f"{symbol} gap {'up' if row['open'] > df.iloc[i-1]['close'] else 'down'}",
                        description=f"Gap of {gap:.5f} detected at open",
                        impact="MEDIUM",
                        affected_instruments=[symbol],
                        data={"gap_size": round(gap, 6)},
                        tags=["gap"],
                    ))

            # New high/low detection (20-bar)
            if row["high"] == df["high"].iloc[max(0, i-20):i+1].max():
                events.append(TimelineEvent(
                    timestamp=df.index[i],
                    source="price",
                    event_type="new_high",
                    title=f"{symbol} 20-bar high",
                    description=f"New 20-bar high at {row['high']:.5f}",
                    impact="LOW",
                    affected_instruments=[symbol],
                    data={"price": round(row["high"], 6)},
                    tags=["breakout", "high"],
                ))

        return events

    @staticmethod
    def from_news(news_items: list[dict]) -> list[TimelineEvent]:
        """Convert news items into timeline events."""
        events = []
        for item in news_items:
            events.append(TimelineEvent(
                timestamp=datetime.fromisoformat(item.get("datetime", datetime.utcnow().isoformat())),
                source="news" if "rate" not in item.get("name", "").lower() else "economic",
                event_type=item.get("type", "news_release"),
                title=item.get("name", item.get("headline", "Unknown")),
                description=item.get("description", ""),
                impact=item.get("impact", "MEDIUM"),
                affected_instruments=item.get("affected_pairs", []),
                data=item,
                tags=item.get("tags", ["news"]),
            ))
        return events

    @staticmethod
    def from_correlation_shift(shifts: list[dict]) -> list[TimelineEvent]:
        """Convert correlation regime shifts into timeline events."""
        events = []
        for shift in shifts:
            events.append(TimelineEvent(
                timestamp=datetime.utcnow(),
                source="correlation",
                event_type="correlation_regime_shift",
                title=f"Correlation shift: {shift['pair']}",
                description=f"Deviation: {shift['deviation']:.4f} — {shift['signal']}",
                impact="HIGH" if abs(shift["deviation"]) > 0.4 else "MEDIUM",
                affected_instruments=shift["pair"].split("/"),
                data=shift,
                tags=["correlation", "regime_shift"],
            ))
        return events

    @staticmethod
    def from_institutional(actions: list[dict]) -> list[TimelineEvent]:
        """Convert institutional actions into timeline events."""
        events = []
        for action in actions:
            events.append(TimelineEvent(
                timestamp=datetime.fromisoformat(action.get("datetime", 

…(truncated)