Institutional Behavior Monitor & Event Timeline Linker
Part 1: Central Bank Tracker
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)
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
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
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:
clusters[-1].append(event)
else:
clusters.append([event])
return clusters
class CausalChainBuilder:
"""Build cause-effect chains from temporally aligned events."""
CAUSAL_PATTERNS = {
("economic", "price"): {"mechanism": "Economic data release triggered market repricing", "base_confidence": 0.7},
("news", "price"): {"mechanism": "News headline drove sentiment shift and price reaction", "base_confidence": 0.5},
("institutional", "price"):{"mechanism": "Institutional flow caused directional pressure", "base_confidence": 0.6},
("geopolitical", "price"): {"mechanism": "Geopolitical event triggered risk repricing", "base_confidence": 0.6},
}
@staticmethod
def build_narrative(cluster: list[TimelineEvent]) -> dict:
"""Build WHAT → WHY → HOW → RESULT → PREDICTION narrative."""
cluster.sort(key=lambda e: e.timestamp)
causes = [e for e in cluster if e.source in ("economic", "news", "institutional", "geopolitical")]
effects = [e for e in cluster if e.source in ("price", "correlation", "technical")]
PREDICTIONS = {
"economic": "Watch for follow-through in direction of surprise. Elevated vol for 2-4h post-release.",
"news": "Sentiment moves often retrace 50-70%. Watch for confirmation at key S/R.",
"institutional":"Institutional flows tend to persist. Watch continuation over 1-3 sessions.",
"geopolitical":"Risk-off flows initially. JPY, CHF, Gold strengthen. Duration depends on escalation.",
}
primary_source = causes[0].source if causes else "unknown"
return {
"time_range": f"{cluster[0].timestamp.isoformat()} → {cluster[-1].timestamp.isoformat()}",
"what": " → ".join([e.title for e in cluster]),
"why": causes[0].description if causes else "Cause unclear — further analysis needed",
"how": "; ".join([f"{e.source}: {e.title}" for e in causes]) if causes else "No clear trigger",
"result": "; ".join([e.title for e in effects]) if effects else "Price impact not yet measured",
"prediction": PREDICTIONS.get(primary_source, "Monitor for follow-through or reversal signals."),
"n_events": len(cluster),
"sources": list(set(e.source for e in cluster)),
}
Timeline Usage Rules
- Always collect from multiple sources — single-source narratives are unreliable
- 4h window for intraday events, 24h for daily, 1 week for macro
- Confidence < 0.5 = hypothesis only, not conclusion
- Predictions are probabilistic — never present as certainties
mahmoud20138