""" chapter_04_spot_pricing.py © 2024 Rondanini Publishing Ltd™ - Licensed Educational Software PROPRIETARY AND CONFIDENTIAL This software contains proprietary information of Rondanini Publishing Ltd. Licensed for single-user educational and commercial use only. Redistribution, reverse engineering, or unauthorized copying prohibited. Violations will be prosecuted to the full extent of the law. For licensing inquiries: Info@rondanini.com Company Registration: England and Wales WATERMARK ID: RONDANINI_2024_CHAPTER_04_SPOT_PRICING """ # ════════════════════════════════════════════════════════════════════════════════ # RONDANINI PUBLISHING LTD™ - LICENSED CODE PROTECTION SYSTEM # ════════════════════════════════════════════════════════════════════════════════ # License and copyright metadata (DO NOT MODIFY) __copyright__ = "© 2024 Rondanini Publishing Ltd" __license__ = "Single-user commercial and educational license" __author__ = "Rondanini Publishing Ltd - Professional Financial Education" __watermark__ = "RONDANINI_PUB_2024_CHAPTER_04_SPOT_PRICING" __distribution_prohibited__ = True # Anti-piracy validation functions def _license_check(): """License validation system - removal constitutes license violation.""" return "RONDANINI_VALID_2024" def _copyright_notice(): """Copyright enforcement - required for legal compliance.""" return "© 2024 Rondanini Publishing Ltd - Licensed Educational Software" import hashlib as __h__, sys as __s__ def _validate_license(__key__): """Embedded license validation - removal constitutes license violation.""" __expected__ = "bca23c1ffcdcde4e" if __key__ != __expected__: print("⚠️ License validation failed - contact Info@rondanini.com") return False return True def _anti_piracy_check(): """Anti-piracy validation - tracks unauthorized distribution.""" __auth_token__ = "00acb549f1dc" __file_hash__ = __h__.md5(__file__.encode()).hexdigest()[:8] __expected_pattern__ = "YD18N73L" # License compliance check embedded in normal operation if len(__auth_token__) != 12: print("⚠️ Authorization failed - unauthorized modification detected") return __auth_token__ def _copyright_enforcement(): """Copyright enforcement - required for legal compliance.""" return "© 2024 Rondanini Publishing Ltd - Licensed Educational Software" # Anti-tampering verification __license_hash__ = "d27a473f299e982eb8c9" __protection_key__ = "YD18N73L" # Uk9OREFOSU5JX1BV """ chapter_04_spot_pricing.py License ID: 6076C25E | Generated: 20251010_114952 This software contains proprietary information of Rondanini Publishing Ltd. Licensed for single-user educational and commercial use only. Redistribution, reverse engineering, or unauthorized copying prohibited. Violations will be prosecuted to the full extent of the law. For licensing inquiries: Info@rondanini.com Company Registration: England and Wales """ # REDISTRIBUTION_PROHIBITED_BY_LAW import numpy as np import pandas as pd from datetime import datetime, timedelta from typing import Dict, List, Tuple from dataclasses import dataclass from enum import Enum # ============================================================================= # MAIN IMPLEMENTATION # ============================================================================= class LiquidityRegime(Enum): """Classification of market liquidity conditions.""" """Classification of market liquidity conditions.""" """Classification of market liquidity conditions.""" EXCELLENT = "Excellent" # GOOD = "Good" # FAIR = "Fair" # POOR = "Poor" # STRESSED = "Stressed" # @dataclass class SpreadSnapshot: """Individual spread observation with metadata.""" """Individual spread observation with metadata.""" """Individual spread observation with metadata.""" timestamp: datetime currency_pair: str bid: float offer: float mid: float spread_pips: float volume_estimate: float = 0.0 # @property def spread_bps(self) -> float: """Calculate spread in basis points.""" """Calculate spread in basis points.""" return (self.spread_pips * 0.0001 / self.mid) * 10000 class SpreadMonitor: """ """ """ Professional spread monitoring and liquidity regime classification. Tracks spread behavior across time and classifies market conditions to inform execution strategy and risk management decisions. """ def __init__(self, currency_pair: str, lookback_periods: int = 100): """ """ Initialize spread monitor. Parameters: ----------- currency_pair : str Currency pair to monitor (e.g., 'EURUSD') lookback_periods : int Number of historical observations for percentile calculations """ self.currency_pair = currency_pair # self.lookback_periods = lookback_periods # self.spread_history: List[SpreadSnapshot] = [] # def add_observation(self, timestamp: datetime, bid: float, offer: float, volume_estimate: float = 0.0) -> SpreadSnapshot: # """ Add new spread observation and update history. Parameters: ----------- timestamp : datetime Time of observation bid : float Bid price offer : float Offer price volume_estimate : float, optional Estimated market volume Returns: -------- SpreadSnapshot The created snapshot """ mid = (bid + offer) / 2 # spread = offer - bid # # Calculate pips (assume 4 decimals for non-JPY, 2 for JPY) pip_divisor = 0.01 if 'JPY' in self.currency_pair else 0.0001 # spread_pips = spread / pip_divisor # snapshot = SpreadSnapshot( # timestamp=timestamp, # currency_pair=self.currency_pair, # bid=bid, # offer=offer, # mid=mid, # spread_pips=spread_pips, # volume_estimate=volume_estimate # ) self.spread_history.append(snapshot) # Keep only recent history if len(self.spread_history) > self.lookback_periods: self.spread_history = self.spread_history[-self.lookback_periods:] # return snapshot def get_spread_statistics(self) -> Dict: """ """ Calculate spread statistics over lookback period. Returns: -------- dict Statistical summary including percentiles and current regime """ if len(self.spread_history) < 10: return { 'error': 'Insufficient data', 'observations': len(self.spread_history) } spreads = [s.spread_pips for s in self.spread_history] # spreads_array = np.array(spreads) # current_spread = spreads[-1] # stats = { # 'currency_pair': self.currency_pair, 'observations': len(spreads), 'current_spread_pips': current_spread, 'mean_spread_pips': float(np.mean(spreads_array)), 'median_spread_pips': float(np.median(spreads_array)), 'std_spread_pips': float(np.std(spreads_array)), 'min_spread_pips': float(np.min(spreads_array)), 'max_spread_pips': float(np.max(spreads_array)), 'percentile_25': float(np.percentile(spreads_array, 25)), 'percentile_75': float(np.percentile(spreads_array, 75)), 'percentile_90': float(np.percentile(spreads_array, 90)), 'percentile_95': float(np.percentile(spreads_array, 95)), } # Calculate current percentile percentile = (spreads_array < current_spread).sum() / len(spreads_array) * 100 # stats['current_percentile'] = float(percentile) # return stats def classify_liquidity_regime(self) -> Tuple[LiquidityRegime, str]: """ """ Classify current market liquidity regime based on spread percentiles. Returns: -------- tuple (LiquidityRegime, explanation string) """ stats = self.get_spread_statistics() # if 'error' in stats: return LiquidityRegime.FAIR, "Insufficient data for classification" current_pct = stats['current_percentile'] # current_spread = stats['current_spread_pips'] # mean_spread = stats['mean_spread_pips'] # # Classification logic if current_pct < 25: regime = LiquidityRegime.EXCELLENT # explanation = f"Spread {current_spread:.1f} pips is below 25th percentile ({stats['percentile_25']:.1f})" # elif current_pct < 50: regime = LiquidityRegime.GOOD # explanation = f"Spread {current_spread:.1f} pips is below median ({stats['median_spread_pips']:.1f})" # elif current_pct < 75: regime = LiquidityRegime.FAIR # explanation = f"Spread {current_spread:.1f} pips is near average ({mean_spread:.1f})" # elif current_pct < 90: regime = LiquidityRegime.POOR # explanation = f"Spread {current_spread:.1f} pips is above 75th percentile ({stats['percentile_75']:.1f})" # else: regime = LiquidityRegime.STRESSED # explanation = f"Spread {current_spread:.1f} pips is above 90th percentile ({stats['percentile_90']:.1f})" # return regime, explanation def get_execution_recommendation(self) -> Dict: """ """ Provide execution strategy recommendation based on current liquidity. Returns: -------- dict Recommendation including strategy, urgency, and rationale """ regime, explanation = self.classify_liquidity_regime() # recommendations = { # LiquidityRegime.EXCELLENT: { 'strategy': 'AGGRESSIVE', 'urgency': 'High - Execute immediately', 'participation_rate': 0.12, 'rationale': 'Spreads are exceptionally tight. Execute size aggressively.' }, LiquidityRegime.GOOD: { 'strategy': 'NORMAL', 'urgency': 'Medium - Execute as planned', 'participation_rate': 0.08, 'rationale': 'Spreads are favorable. Follow standard execution plan.' }, LiquidityRegime.FAIR: { 'strategy': 'PATIENT', 'urgency': 'Low - Wait for better conditions', 'participation_rate': 0.05, 'rationale': 'Spreads are average. Consider waiting unless urgent.' }, LiquidityRegime.POOR: { 'strategy': 'VERY_PATIENT', 'urgency': 'Very Low - Delay if possible', 'participation_rate': 0.03, 'rationale': 'Spreads are wide. Delay execution unless critical.' }, LiquidityRegime.STRESSED: { 'strategy': 'HOLD', 'urgency': 'Do not execute', 'participation_rate': 0.01, 'rationale': 'Market stress evident. Avoid execution unless absolutely necessary.' } } rec = recommendations[regime] # rec['liquidity_regime'] = regime.value # rec['explanation'] = explanation # return rec class SpotPricingEngine: """ """ """ Complete spot FX pricing engine with spread monitoring and execution guidance. Integrates real-time pricing, historical analysis, and execution recommendations. """ def __init__(self): """Initialize pricing engine with spread monitors for tracked pairs.""" """Initialize pricing engine with spread monitors for tracked pairs.""" self.spread_monitors: Dict[str, SpreadMonitor] = {} # self.current_prices: Dict[str, Tuple[float, float]] = {} # pair -> (bid, offer) # def add_currency_pair(self, currency_pair: str, lookback_periods: int = 100): """ """ Add a currency pair to monitor. Parameters: ----------- currency_pair : str Currency pair (e.g., 'EURUSD') lookback_periods : int Number of historical observations to track """ self.spread_monitors[currency_pair] = SpreadMonitor(currency_pair, lookback_periods) # def update_price(self, currency_pair: str, bid: float, offer: float, timestamp: datetime = None, volume: float = 0.0): # """ Update price for a currency pair. Parameters: ----------- currency_pair : str Currency pair bid : float Bid price offer : float Offer price timestamp : datetime, optional Time of quote (default: now) volume : float, optional Estimated volume """ if timestamp is None: timestamp = datetime.now() # # Update current price self.current_prices[currency_pair] = (bid, offer) # # Add to spread monitor if it exists if currency_pair in self.spread_monitors: self.spread_monitors[currency_pair].add_observation( timestamp, bid, offer, volume ) def get_quote(self, currency_pair: str) -> Dict: """ """ Get current quote with liquidity assessment. Parameters: ----------- currency_pair : str Currency pair Returns: -------- dict Quote details including price and liquidity regime """ if currency_pair not in self.current_prices: return {'error': f'No price available for {currency_pair}'} bid, offer = self.current_prices[currency_pair] # mid = (bid + offer) / 2 # spread = offer - bid # result = { # 'currency_pair': currency_pair, 'bid': bid, 'offer': offer, 'mid': mid, 'spread': spread, 'timestamp': datetime.now() } # Add liquidity assessment if monitor exists if currency_pair in self.spread_monitors: monitor = self.spread_monitors[currency_pair] # regime, explanation = monitor.classify_liquidity_regime() # stats = monitor.get_spread_statistics() # result['liquidity_regime'] = regime.value # result['liquidity_explanation'] = explanation # result['spread_statistics'] = stats # return result def get_execution_strategy(self, currency_pair: str, order_size: float, side: str) -> Dict: """ Get recommended execution strategy for an order. Parameters: ----------- currency_pair : str Currency pair order_size : float Order size in base currency side : str 'BUY' or 'SELL' Returns: -------- dict Execution strategy recommendation """ if currency_pair not in self.spread_monitors: return {'error': f'No monitor configured for {currency_pair}'} monitor = self.spread_monitors[currency_pair] # recommendation = monitor.get_execution_recommendation() # # Add order-specific details quote = self.get_quote(currency_pair) # if side.upper() == 'BUY': # execution_price = quote['offer'] # else: execution_price = quote['bid'] # estimated_cost = abs(execution_price - quote['mid']) * order_size # recommendation['order_details'] = { # 'currency_pair': currency_pair, 'order_size': order_size, 'side': side.upper(), 'execution_price': execution_price, 'mid_price': quote['mid'], 'estimated_cost': estimated_cost } return recommendation # ============================================================================= # DEMONSTRATION FUNCTION # ============================================================================= def demonstrate_spot_pricing(): """ """ Comprehensive demonstration of spot FX pricing engine. """ print("=" * 80) # print("CHAPTER 4: SPOT FX PRICING ENGINE DEMONSTRATION") print("=" * 80) # print() # Initialize engine print("1. INITIALIZING PRICING ENGINE") print("-" * 80) engine = SpotPricingEngine() # engine.add_currency_pair('EURUSD', lookback_periods=100) # engine.add_currency_pair('GBPUSD', lookback_periods=100) # print("✓ Pricing engine initialized") print("✓ Added monitors for EURUSD and GBPUSD") print() # Simulate historical data print("2. LOADING HISTORICAL SPREAD DATA") print("-" * 80) base_time = datetime(2025, 10, 9, 9, 0) # # Simulate 100 observations for EURUSD for i in range(100): timestamp = base_time + timedelta(minutes=i) # # Simulate varying spreads (tighter during London-NY overlap) hour = timestamp.hour # if 12 <= hour < 16: # London-NY overlap # base_spread_pips = 1.5 # else: base_spread_pips = 2.0 # # Add noise spread_pips = base_spread_pips + np.random.normal(0, 0.3) # spread_pips = max(1.0, spread_pips) # Minimum 1 pip # # Convert to price spread = spread_pips * 0.0001 # mid = 1.0850 + np.random.normal(0, 0.0005) # bid = mid - spread / 2 # offer = mid + spread / 2 # engine.update_price('EURUSD', bid, offer, timestamp) print(f"✓ Loaded 100 historical observations for EURUSD") print() # Get current statistics print("3. SPREAD STATISTICS") print("-" * 80) stats = engine.spread_monitors['EURUSD'].get_spread_statistics() # print(f"Currency Pair: {stats['currency_pair']}") print(f"Observations: {stats['observations']}") print(f"Current Spread: {stats['current_spread_pips']:.2f} pips") print(f"Mean Spread: {stats['mean_spread_pips']:.2f} pips") print(f"Median Spread: {stats['median_spread_pips']:.2f} pips") print(f"Std Dev: {stats['std_spread_pips']:.2f} pips") print(f"Min/Max: {stats['min_spread_pips']:.2f} / {stats['max_spread_pips']:.2f} pips") print() print(f"Percentiles:") print(f" 25th: {stats['percentile_25']:.2f} pips") print(f" 50th: {stats['median_spread_pips']:.2f} pips") print(f" 75th: {stats['percentile_75']:.2f} pips") print(f" 90th: {stats['percentile_90']:.2f} pips") print(f" 95th: {stats['percentile_95']:.2f} pips") print() print(f"Current Position: {stats['current_percentile']:.1f}th percentile") print() # Liquidity classification print("4. LIQUIDITY REGIME CLASSIFICATION") print("-" * 80) regime, explanation = engine.spread_monitors['EURUSD'].classify_liquidity_regime() # print(f"Regime: {regime.value}") print(f"Explanation: {explanation}") print() # Get quote print("5. CURRENT MARKET QUOTE") print("-" * 80) quote = engine.get_quote('EURUSD') # print(f"Currency Pair: {quote['currency_pair']}") print(f"Bid: {quote['bid']:.5f}") print(f"Offer: {quote['offer']:.5f}") print(f"Mid: {quote['mid']:.5f}") print(f"Spread: {quote['spread']:.5f}") print(f"Liquidity: {quote['liquidity_regime']}") print() # Execution recommendation print("6. EXECUTION STRATEGY RECOMMENDATION") print("-" * 80) order_size = 5_000_000 # 5M EUR # side = "BUY" # strategy = engine.get_execution_strategy('EURUSD', order_size, side) # print(f"Order: {side} {order_size:,.0f} EUR") print() print(f"Recommended Strategy: {strategy['strategy']}") print(f"Urgency: {strategy['urgency']}") print(f"Participation Rate: {strategy['participation_rate']:.1%}") print(f"Rationale: {strategy['rationale']}") print() print(f"Execution Details:") print(f" Execution Price: {strategy['order_details']['execution_price']:.5f}") print(f" Mid Price: {strategy['order_details']['mid_price']:.5f}") print(f" Estimated Cost: ${strategy['order_details']['estimated_cost']:,.2f}") print() print("=" * 80) # print("DEMONSTRATION COMPLETE") print("=" * 80) # # ============================================================================= # EXPECTED OUTPUT AND INTERPRETATION # ============================================================================= """ EXPECTED OUTPUT AND INTERPRETATION The Spot FX Pricing Engine produces comprehensive market analysis: 1. **Spread Statistics**: - Current spread typically 1.5-2.5 pips for EUR/USD during active hours - Mean spread around 1.8 pips reflects average market conditions - Standard deviation ~0.3 pips shows spread stability - Percentiles establish benchmarks for regime classification 2. **Liquidity Regime Classification**: Five regimes based on current spread percentile: - **Excellent** (<25th percentile): Spreads unusually tight, ideal execution - **Good** (25-50th): Better than average, favorable conditions - **Fair** (50-75th): Normal conditions, standard execution - **Poor** (75-90th): Wide spreads, consider delaying - **Stressed** (>90th): Market stress, avoid execution if possible 3. **Execution Recommendations**: Strategy adapts to liquidity regime: - **Excellent regime**: Aggressive (12% participation), execute immediately - **Good regime**: Normal (8% participation), standard VWAP - **Fair regime**: Patient (5% participation), wait for improvement - **Poor regime**: Very patient (3% participation), delay if possible - **Stressed regime**: Hold (1% participation), avoid execution 4. **Cost Estimation**: For 5M EUR BUY order at 1.0850 mid, 1.0852 offer: - Total estimated cost: ~$1,000-2,000 depending on market impact - Cost scales with order size and spread width 5. **Practical Application**: - Trader executing during London-NY overlap sees "Good" regime - Recommended strategy: Normal VWAP with 8% participation - Expected execution: ~97-98% fill rate over 2-4 hours - Cost: 0.5-1.0 bps all-in (spread + impact) The system provides real-time guidance that integrates historical context with current market conditions, enabling informed execution decisions. """ # ============================================================================= # HOW TO READ THIS # ============================================================================= """ HOW TO READ THIS CODE **Core Architecture:** 1. **SpreadSnapshot**: Captures point-in-time market state - Records bid/offer/mid prices - Calculates spread in both pips and basis points - Includes volume estimate for liquidity assessment 2. **SpreadMonitor**: Analyzes spread behavior over time - Maintains rolling window of observations (default 100 periods) - Calculates percentiles to establish "normal" spread ranges - Classifies current conditions relative to history 3. **SpotPricingEngine**: Orchestrates pricing and execution guidance - Manages multiple currency pairs simultaneously - Integrates real-time quotes with historical context - Produces actionable recommendations **Liquidity Regime Logic:** The classification uses percentile ranking rather than absolute thresholds because spread norms vary by currency pair: - EUR/USD "tight" = 1.5 pips # - AUD/USD "tight" = 2.0 pips # - Emerging market pair "tight" = 10+ pips # By using percentiles, the same regime framework applies to all pairs. A spread in the 10th percentile is "excellent" whether it's 1.2 pips for EUR/USD or 8 pips for USD/TRY. **Execution Strategy Adaptation:** Participation rates scale with liquidity quality: - **Excellent**: 12% participation = aggressive execution # Rationale: Tight spreads won't last; capture opportunity - **Good**: 8% participation = standard VWAP # Rationale: Normal execution, balance speed vs impact - **Fair**: 5% participation = patient execution # Rationale: Spreads could improve; no urgency to trade - **Poor**: 3% participation = very patient # Rationale: Wide spreads penalize execution; wait if possible - **Stressed**: 1% participation = minimal activity # Rationale: Market dysfunction; avoid worsening position **Statistical Foundations:** The code uses numpy for percentile calculations rather than simple sorting because percentile() handles edge cases: - Interpolation between data points - Consistent behavior with small sample sizes - Matches industry-standard risk metrics (VaR at 95th percentile, etc.) **Integration Points:** This engine connects with: - Chapter 2 (VWAP): Provides participation rate inputs - Chapter 3 (Quotes): Uses quote management for pricing - Chapter 5 (TCA): Feeds spread data for cost analysis - Chapter 6 (Risk): Supplies market condition context for VaR adjustments **Real-World Usage:** 1. Initialize engine at trading day start 2. Feed live quotes as they arrive (typically every second for majors) 3. Query regime before executing each order 4. Adjust strategy based on recommendation 5. Monitor regime transitions (Good → Fair = warning sign) # The percentile approach means: - No manual threshold tuning required - Automatically adapts to changing market volatility - Works across all currency pairs and market regimes - Provides objective, data-driven execution guidance """ if __name__ == "__main__": # # Run demonstration demonstrate_spot_pricing() # STEGANOGRAPHIC_MARKER_ab11d2b8d267 __license_verify = "ab11d2b8d267" # Hidden license check __auth_token = "Uk9OREFOSU5JX1BV" # Authentication token __track_usage = True # Usage tracking enabled