institutional-timeline

# Institutional Behavior Monitor & Event Timeline Linker ## Part 1: Central Bank Tracker ```python from dataclasses import dataclass, field from datetime import datetime, timedelta from typing import Optional, Literal import pandas as pd import numpy as np @dataclass class CentralBankProfile: name: str code: str currency: str current_rate: float last_decision: str # "hike", "cut", "hold" last_decision_date: str next_meeting: str bias: str # "hawkish", "dovish", "neutral" qe_status: str # "tightening", "stable", "expanding" key_officials: list[str] affected_pairs: list[str] CENTRAL_BANKS = { "FED": CentralBankProfile("Federal Reserve", "FED", "USD", 0.0, "hold", "", "", "neutral", "tightening", ["Chair","Vice Chair"], ["EURUSD","USDJPY","GBPUSD","USDCHF","AUDUSD","USDCAD","NZDUSD","XAUUSD"]), "ECB": CentralBankProfile("European Central Bank", "ECB", "EUR", 0.0, "hold", "", "", "neutral", "stable", ["President","Chief Economist"], ["EURUSD","EURJPY","EURGBP","EURAUD","EURCHF"]), "BOE": CentralBankProfile("Bank of England", "BOE", "GBP", 0.0, "hold", "", "", "neutral", "stable", ["Governor","Deputy Governor"], ["GBPUSD","EURGBP","GBPJPY"]), "BOJ": CentralBankProfile("Bank of Japan", "BOJ", "JPY", 0.0, "hold", "", "", "dovish", "expanding", ["Governor","Deputy Governor"], ["USDJPY","EURJPY","GBPJPY","AUDJPY"]), "RBA": CentralBankProfile("Reserve Bank of Aus", "RBA", "AUD", 0.0, "hold", "", "", "neutral", "stable", ["Governor","Deputy Governor"], ["AUDUSD","AUDNZD","EURAUD","AUDJPY"]), "BOC": CentralBankProfile("Bank of Canada", "BOC", "CAD", 0.0, "hold", "", "", "neutral", "stable", ["Governor","Senior Deputy"], ["USDCAD","CADJPY"]), "SNB": CentralBankProfile("Swiss National Bank", "SNB", "CHF", 0.0, "hold", "", "", "neutral", "stable", ["Chairman","Vice Chairman"], ["USDCHF","EURCHF"]), "RBNZ": CentralBankProfile("Reserve Bank of NZ", "RBNZ", "NZD", 0.0, "hold", "", "", "neutral", "stable", ["Governor","Deputy Governor"], ["NZDUSD","AUDNZD"]), } class CentralBankTracker: def get_policy_divergence(self) -> pd.DataFrame: """Map policy divergence — divergence drives currency pair trends.""" data = [] banks = list(CENTRAL_BANKS.values()) bias_map = {"hawkish": 1, "neutral": 0, "dovish": -1} for i, bank_a in enumerate(banks): for bank_b in banks[i + 1:]: rate_diff = bank_a.current_rate - bank_b.current_rate bias_diff = bias_map.get(bank_a.bias, 0) - bias_map.get(bank_b.bias, 0) data.append({ "pair": f"{bank_a.currency}/{bank_b.currency}", "rate_differential": round(rate_diff, 2), "bias_differential": bias_diff, "divergence_signal": self._interpret_divergence(rate_diff, bias_diff), }) return pd.DataFrame(data) @staticmethod def _interpret_divergence(rate_diff: float, bias_diff: int) -> str: if rate_diff > 0.5 and bias_diff > 0: return "STRONG BUY base — widening rate + hawkish bias" if rate_diff < -0.5 and bias_diff < 0: return "STRONG SELL base — widening rate disadvantage" if abs(rate_diff) < 0.25 and bias_diff == 0: return "NEUTRAL — no clear divergence" if bias_diff > 0: return "MILD BUY base — bias divergence" if bias_diff < 0: return "MILD SELL base — quote currency favored" return "MIXED — conflicting signals" @staticmethod def rate_decision_impact_model(code: str, decision: str, expected: str) -> dict: """Model market impact of rate decision vs expectations.""" bank = CENTRAL_BANKS.get(code) if not bank: return {"error": f"Unknown bank: {code}"} SURPRISE_MAP = { ("hike", "hold"): {"direction": "bullish", "magnitude": "large", "confidence": 0.85}, ("hike", "hike"): {"direction": "neutral_to_bullish","magnitude": "small", "confidence": 0.50}, ("cut", "hold"): {"direction": "bearish", "magnitude": "large", "confidence": 0.85}, ("cut", "cut"): {"direction": "neutral_to_bearish","magnitude": "small", "confidence": 0.50}, ("hold", "hike"): {"direction": "bearish", "magnitude": "medium", "confidence": 0.70}, ("hold", "cut"): {"direction": "bullish", "magnitude": "medium", "confidence": 0.70}, ("hold", "hold"): {"direction": "neutral", "magnitude": "minimal", "confidence": 0.30}, } impact = SURPRISE_MAP.get((decision.lower(), expected.lower()), {"direction": "unknown", "magnitude": "unknown", "confidence": 0}) return { "bank": bank.name, "currency": bank.currency, "decision": decision, "expected": expected, "is_surprise": decision != expected, "impact": impact, "affected_pairs": bank.affected_pairs, "recommendation": f"{'Strong' if impact['magnitude'] == 'large' else 'Moderate'} " f"{impact['direction']} signal on {bank.currency} pairs" if impact["confidence"] > 0.5 else "Wait for price action confirmation", } ``` --- ## Part 2: COT Analyzer (Full Institutional) ```python class COTAnalyzer: CURRENCY_CONTRACTS = { "EUR": "EURO FX", "GBP": "BRITISH POUND", "JPY": "JAPANESE YEN", "AUD": "AUSTRALIAN DOLLAR", "CAD": "CANADIAN DOLLAR", "CHF": "SWISS FRANC", "NZD": "NEW ZEALAND DOLLAR", "XAU": "GOLD", "WTI": "CRUDE OIL", } @staticmethod def analyze_positioning(cot_data: pd.DataFrame, currency: str) -> dict: """ cot_data columns: [date, long_noncommercial, short_noncommercial, ...] """ latest = cot_data.iloc[-1] prev = cot_data.iloc[-2] if len(cot_data) > 1 else latest net_spec = latest["long_noncommercial"] - latest["short_noncommercial"] prev_net_spec = prev["long_noncommercial"] - prev["short_noncommercial"] net_change = net_spec - prev_net_spec historical_net = cot_data["long_noncommercial"] - cot_data["short_noncommercial"] percentile = (historical_net < net_spec).mean() * 100 signal = ( "EXTREME LONG + UNWINDING → potential reversal (bearish)" if percentile > 90 and net_change < 0 else "EXTREME SHORT + COVERING → potential reversal (bullish)" if percentile < 10 and net_change > 0 else "STRONG LONG + BUILDING → continuation but watch crowding" if percentile > 80 else "MODERATE positioning — no extreme signal" ) return { "currency": currency, "net_speculative": int(net_spec), "weekly_change": int(net_change), "direction": "LONG" if net_spec > 0 else "SHORT", "historical_percentile": round(percentile, 1), "extreme_positioning": percentile > 90 or percentile < 10, "signal": signal, } ``` --- ## Part 3: Intervention Detector ```python class InterventionDetector: @staticmethod def detect_fx_intervention(df: pd.DataFrame, symbol: str, atr_multiplier: float = 5.0, volume_multiplier: float = 3.0) -> list[dict]: """Detect abnormal price moves indicating possible central bank intervention.""" atr_val = (df["high"] - df["low"]).rolling(14).mean() vol_avg = df["volume"].rolling(20).mean() detections = [] for i in range(20, len(df)): bar_range = df.iloc[i]["high"] - df.iloc[i]["low"] bar_vol = df.iloc[i]["volume"] if bar_range > atr_multiplier * atr_val.iloc[i] and bar_vol > volume_multiplier * vol_avg.iloc[i]: prior_trend = df["close"].iloc[i-10:i].pct_change().mean() bar_direction = 1 if df.iloc[i]["close"] > df.iloc[i]["open"] else -1 is_reversal = (prior_trend > 0 and bar_direction < 0) or (prior_trend < 0 and bar_direction > 0) detections.append({ "timestamp": df.index[i].isoformat(), "symbol": symbol, "bar_range_atr": round(bar_range / atr_val.iloc[i], 2), "volume_ratio": round(bar_vol / vol_avg.iloc[i], 2), "is_trend_reversal": is_reversal, "intervention_probability": round( min(0.3 + (bar_range / atr_val.iloc[i]) * 0.1 + (0.2 if is_reversal else 0), 0.95), 3), "note": "Possible central bank intervention" if is_reversal else "Large institutional flow", }) return detections @staticmethod def web_search_queries(): return [ 'web_search("Federal Reserve interest rate 2025 current")', 'web_search("ECB monetary policy latest decision")', 'web_search("Goldman Sachs EURUSD forecast")', 'web_search("JPMorgan FX outlook currencies")', 'web_search("CFTC COT report latest forex positioning")'] ``` --- ## Part 4: Event Timeline Linker ```python from dataclasses import dataclass, field, asdict @dataclass class TimelineEvent: """Universal event structure for cross-source linking.""" timestamp: datetime source: Literal["price", "news", "economic", "institutional", "correlation", "technical", "geopolitical"] event_type: str title: str description: str impact: Literal["HIGH", "MEDIUM", "LOW"] affected_instruments: list[str] = field(default_factory=list) data: dict = field(default_factory=dict) tags: list[str] = field(default_factory=list) class TemporalAligner: @staticmethod def find_concurrent_events(events: list[TimelineEvent], target_time: datetime, window: timedelta = timedelta(hours=4)) -> list[TimelineEvent]: return [e for e in events if abs((e.timestamp - target_time).total_seconds()) <= window.total_seconds()] @staticmethod def cluster_by_time(events: list[TimelineEvent], max_gap: timedelta = timedelta(hours=2)) -> list[list[TimelineEvent]]: if not events: return [] sorted_events = sorted(events, key=lambda e: e.timestamp) clusters = [[sorted_events[0]]] for event in sorted_events[1:]: if event.timestamp - clusters[-1][-1].timestamp <= max_gap: