45. Adaptive Trading Frequency Controller¶

services/adaptive-frequency-center/
├── src/
│   ├── main.py
│   ├── monitor/
│   │   ├── market_state_monitor.py
│   │   ├── volatility_tracker.py
│   │   └── liquidity_monitor.py
│   ├── decision/
│   │   ├── frequency_decider.py
│   │   ├── mode_selector.py
│   │   └── threshold_manager.py
│   ├── controller/
│   │   ├── schedule_controller.py
│   │   ├── event_scheduler.py
│   │   └── frequency_optimizer.py
│   ├── adapter/
│   │   ├── execution_rate_limiter.py
│   │   ├── strategy_adapter.py
│   │   └── signal_throttler.py
│   ├── cost/
│   │   ├── transaction_cost_analyzer.py
│   │   └── slippage_predictor.py
│   ├── api/
│   │   ├── frequency_api.py
│   ├── config.py
│   └── requirements.txt
├── Dockerfile
└── tests/

import numpy as np
from typing import Dict, List
from datetime import datetime, timedelta

class MarketStateMonitor:
    def __init__(self, volatility_window=100, volume_window=30, 
                 price_window=50):
        self.volatility_window = volatility_window
        self.volume_window = volume_window
        self.price_window = price_window
        self.market_data = {}
        self.state_history = {}

    def update_market_data(self, symbol: str, price: float, 
                          volume: float, timestamp: datetime):
        """Update market data for state monitoring"""
        if symbol not in self.market_data:
            self.market_data[symbol] = {
                'prices': [],
                'volumes': [],
                'timestamps': []
            }

        data = self.market_data[symbol]
        data['prices'].append(price)
        data['volumes'].append(volume)
        data['timestamps'].append(timestamp)

        # Keep only recent data
        if len(data['prices']) > max(self.volatility_window, self.volume_window):
            data['prices'] = data['prices'][-self.volatility_window:]
            data['volumes'] = data['volumes'][-self.volume_window:]
            data['timestamps'] = data['timestamps'][-self.volatility_window:]

    def get_market_state(self, symbol: str) -> Dict:
        """Get current market state for a symbol"""
        if symbol not in self.market_data:
            return self.get_default_state()

        data = self.market_data[symbol]

        if len(data['prices']) < 2:
            return self.get_default_state()

        # Calculate volatility
        returns = np.diff(np.log(data['prices']))
        volatility = np.std(returns)

        # Calculate volume metrics
        if len(data['volumes']) > 0:
            avg_volume = np.mean(data['volumes'])
            volume_trend = self.calculate_volume_trend(data['volumes'])
        else:
            avg_volume = 0
            volume_trend = 0

        # Calculate trading density
        trading_density = self.calculate_trading_density(data['timestamps'])

        # Calculate liquidity score
        liquidity_score = self.calculate_liquidity_score(data['volumes'], volatility)

        market_state = {
            "symbol": symbol,
            "timestamp": datetime.now().isoformat(),
            "volatility": volatility,
            "avg_volume": avg_volume,
            "volume_trend": volume_trend,
            "trading_density": trading_density,
            "liquidity_score": liquidity_score,
            "market_regime": self.classify_market_regime(volatility, avg_volume, trading_density)
        }

        # Store state history
        if symbol not in self.state_history:
            self.state_history[symbol] = []
        self.state_history[symbol].append(market_state)

        # Keep only recent history
        if len(self.state_history[symbol]) > 1000:
            self.state_history[symbol] = self.state_history[symbol][-1000:]

        return market_state

    def calculate_volume_trend(self, volumes: List[float]) -> float:
        """Calculate volume trend (positive = increasing, negative = decreasing)"""
        if len(volumes) < 2:
            return 0.0

        # Simple linear trend
        x = np.arange(len(volumes))
        slope = np.polyfit(x, volumes, 1)[0]
        return slope / np.mean(volumes) if np.mean(volumes) > 0 else 0.0

    def calculate_trading_density(self, timestamps: List[datetime]) -> float:
        """Calculate trading density (trades per minute)"""
        if len(timestamps) < 2:
            return 0.0

        # Calculate time span
        time_span = (timestamps[-1] - timestamps[0]).total_seconds() / 60  # minutes
        if time_span <= 0:
            return 0.0

        return len(timestamps) / time_span

    def calculate_liquidity_score(self, volumes: List[float], volatility: float) -> float:
        """Calculate liquidity score (higher = more liquid)"""
        if not volumes:
            return 0.0

        avg_volume = np.mean(volumes)
        # Normalize by volatility (higher volume + lower volatility = higher liquidity)
        liquidity_score = avg_volume / (1 + volatility * 100)
        return liquidity_score

    def classify_market_regime(self, volatility: float, avg_volume: float, 
                             trading_density: float) -> str:
        """Classify current market regime"""
        if volatility > 0.03 and trading_density > 10:
            return "high_volatility_active"
        elif volatility > 0.02:
            return "high_volatility"
        elif trading_density > 5 and avg_volume > 1000:
            return "high_activity"
        elif volatility < 0.01 and trading_density < 2:
            return "low_activity"
        else:
            return "normal"

    def get_default_state(self) -> Dict:
        """Get default market state"""
        return {
            "symbol": "unknown",
            "timestamp": datetime.now().isoformat(),
            "volatility": 0.01,
            "avg_volume": 0.0,
            "volume_trend": 0.0,
            "trading_density": 0.0,
            "liquidity_score": 0.0,
            "market_regime": "normal"
        }

class FrequencyDecider:
    def __init__(self, base_frequency="medium", 
                 volatility_thresholds=(0.01, 0.02, 0.03),
                 volume_thresholds=(500, 1000, 2000),
                 density_thresholds=(2, 5, 10)):
        self.base_frequency = base_frequency
        self.volatility_thresholds = volatility_thresholds
        self.volume_thresholds = volume_thresholds
        self.density_thresholds = density_thresholds
        self.frequency_history = []
        self.current_frequency = base_frequency

    def decide_frequency(self, market_state: Dict) -> str:
        """Decide optimal trading frequency based on market state"""

        volatility = market_state.get("volatility", 0.01)
        avg_volume = market_state.get("avg_volume", 0.0)
        trading_density = market_state.get("trading_density", 0.0)
        liquidity_score = market_state.get("liquidity_score", 0.0)
        market_regime = market_state.get("market_regime", "normal")

        # Calculate frequency score based on multiple factors
        frequency_score = self.calculate_frequency_score(
            volatility, avg_volume, trading_density, liquidity_score
        )

        # Determine frequency based on score
        if frequency_score > 0.7:
            new_frequency = "high"
        elif frequency_score > 0.3:
            new_frequency = "medium"
        else:
            new_frequency = "low"

        # Apply regime-specific adjustments
        new_frequency = self.apply_regime_adjustments(new_frequency, market_regime)

        # Record frequency change
        if new_frequency != self.current_frequency:
            self.record_frequency_change(new_frequency, market_state)

        self.current_frequency = new_frequency
        return new_frequency

    def calculate_frequency_score(self, volatility: float, avg_volume: float,
                                trading_density: float, liquidity_score: float) -> float:
        """Calculate frequency score (0-1) based on market conditions"""

        # Volatility component (higher volatility = higher frequency)
        vol_score = min(volatility / 0.05, 1.0)  # Normalize to 5% volatility

        # Volume component (higher volume = higher frequency)
        vol_norm = min(avg_volume / 5000, 1.0)  # Normalize to 5000 volume

        # Trading density component
        density_score = min(trading_density / 20, 1.0)  # Normalize to 20 trades/min

        # Liquidity component (higher liquidity = higher frequency)
        liquidity_norm = min(liquidity_score / 1000, 1.0)  # Normalize to 1000 liquidity

        # Weighted combination
        frequency_score = (
            vol_score * 0.3 +
            vol_norm * 0.2 +
            density_score * 0.3 +
            liquidity_norm * 0.2
        )

        return frequency_score

    def apply_regime_adjustments(self, base_frequency: str, market_regime: str) -> str:
        """Apply regime-specific frequency adjustments"""

        regime_adjustments = {
            "high_volatility_active": "high",
            "high_volatility": "medium",
            "high_activity": "high",
            "low_activity": "low",
            "normal": base_frequency
        }

        return regime_adjustments.get(market_regime, base_frequency)

    def record_frequency_change(self, new_frequency: str, market_state: Dict):
        """Record frequency change for analysis"""
        change_record = {
            "timestamp": datetime.now().isoformat(),
            "old_frequency": self.current_frequency,
            "new_frequency": new_frequency,
            "market_state": market_state,
            "reason": self.get_change_reason(new_frequency, market_state)
        }

        self.frequency_history.append(change_record)

        # Keep only recent history
        if len(self.frequency_history) > 1000:
            self.frequency_history = self.frequency_history[-1000:]

    def get_change_reason(self, new_frequency: str, market_state: Dict) -> str:
        """Get reason for frequency change"""
        volatility = market_state.get("volatility", 0)
        trading_density = market_state.get("trading_density", 0)

        if new_frequency == "high":
            if volatility > 0.03:
                return "high_volatility"
            elif trading_density > 10:
                return "high_activity"
            else:
                return "market_opportunity"
        elif new_frequency == "low":
            if volatility < 0.01:
                return "low_volatility"
            elif trading_density < 2:
                return "low_activity"
            else:
                return "cost_optimization"
        else:
            return "normal_conditions"

import asyncio
from typing import Callable, Dict, Any

class ScheduleController:
    def __init__(self, frequency_decider, mode="adaptive"):
        self.frequency_decider = frequency_decider
        self.mode = mode  # "fixed", "event_driven", "adaptive"
        self.current_frequency = "medium"
        self.execution_intervals = {
            "high": 1,      # 1 second
            "medium": 60,   # 1 minute
            "low": 300      # 5 minutes
        }
        self.running_strategies = {}
        self.schedule_history = []

    async def schedule_strategy(self, strategy_id: str, 
                              strategy_executor: Callable,
                              market_state: Dict):
        """Schedule strategy execution based on current frequency"""

        # Update frequency decision
        new_frequency = self.frequency_decider.decide_frequency(market_state)

        if new_frequency != self.current_frequency:
            await self.update_schedule(strategy_id, new_frequency)

        # Get execution interval
        interval = self.execution_intervals[new_frequency]

        # Schedule execution
        if self.mode == "adaptive":
            await self.adaptive_schedule(strategy_id, strategy_executor, interval)
        elif self.mode == "event_driven":
            await self.event_driven_schedule(strategy_id, strategy_executor, market_state)
        else:  # fixed
            await self.fixed_schedule(strategy_id, strategy_executor, interval)

    async def adaptive_schedule(self, strategy_id: str, 
                              strategy_executor: Callable, interval: int):
        """Adaptive scheduling with dynamic interval adjustment"""

        while strategy_id in self.running_strategies:
            try:
                # Execute strategy
                await strategy_executor()

                # Record execution
                self.record_execution(strategy_id, interval)

                # Wait for next execution
                await asyncio.sleep(interval)

            except Exception as e:
                # Log error and continue
                print(f"Strategy execution error for {strategy_id}: {e}")
                await asyncio.sleep(interval)

    async def event_driven_schedule(self, strategy_id: str, 
                                  strategy_executor: Callable, market_state: Dict):
        """Event-driven scheduling based on market events"""

        # Check for significant market events
        if self.should_execute_on_event(market_state):
            await strategy_executor()
            self.record_execution(strategy_id, 0)  # Event-driven

    async def fixed_schedule(self, strategy_id: str, 
                           strategy_executor: Callable, interval: int):
        """Fixed interval scheduling"""

        while strategy_id in self.running_strategies:
            try:
                await strategy_executor()
                self.record_execution(strategy_id, interval)
                await asyncio.sleep(interval)
            except Exception as e:
                print(f"Strategy execution error for {strategy_id}: {e}")
                await asyncio.sleep(interval)

    def should_execute_on_event(self, market_state: Dict) -> bool:
        """Determine if strategy should execute based on market events"""

        volatility = market_state.get("volatility", 0)
        trading_density = market_state.get("trading_density", 0)

        # Execute on significant market events
        if volatility > 0.03:  # High volatility event
            return True
        elif trading_density > 15:  # High activity event
            return True
        elif abs(market_state.get("volume_trend", 0)) > 0.5:  # Volume spike
            return True

        return False

    async def update_schedule(self, strategy_id: str, new_frequency: str):
        """Update execution schedule for a strategy"""

        schedule_update = {
            "timestamp": datetime.now().isoformat(),
            "strategy_id": strategy_id,
            "old_frequency": self.current_frequency,
            "new_frequency": new_frequency,
            "old_interval": self.execution_intervals[self.current_frequency],
            "new_interval": self.execution_intervals[new_frequency]
        }

        self.schedule_history.append(schedule_update)
        self.current_frequency = new_frequency

        # Keep only recent history
        if len(self.schedule_history) > 1000:
            self.schedule_history = self.schedule_history[-1000:]

    def record_execution(self, strategy_id: str, interval: int):
        """Record strategy execution"""
        execution_record = {
            "timestamp": datetime.now().isoformat(),
            "strategy_id": strategy_id,
            "frequency": self.current_frequency,
            "interval": interval
        }

        if strategy_id not in self.running_strategies:
            self.running_strategies[strategy_id] = []

        self.running_strategies[strategy_id].append(execution_record)

        # Keep only recent executions
        if len(self.running_strategies[strategy_id]) > 100:
            self.running_strategies[strategy_id] = self.running_strategies[strategy_id][-100:]

    def start_strategy(self, strategy_id: str):
        """Start scheduling for a strategy"""
        self.running_strategies[strategy_id] = []

    def stop_strategy(self, strategy_id: str):
        """Stop scheduling for a strategy"""
        if strategy_id in self.running_strategies:
            del self.running_strategies[strategy_id]

class ExecutionRateLimiter:
    def __init__(self, schedule_controller, max_executions_per_minute=60):
        self.schedule_controller = schedule_controller
        self.max_executions_per_minute = max_executions_per_minute
        self.execution_history = {}
        self.rate_limits = {}

    async def execute_with_rate_limit(self, strategy_id: str, 
                                    strategy_executor: Callable,
                                    market_state: Dict):
        """Execute strategy with rate limiting"""

        # Check rate limits
        if not self.check_rate_limit(strategy_id):
            return False

        # Execute strategy
        await self.schedule_controller.schedule_strategy(
            strategy_id, strategy_executor, market_state
        )

        # Record execution
        self.record_execution(strategy_id)

        return True

    def check_rate_limit(self, strategy_id: str) -> bool:
        """Check if execution is within rate limits"""

        if strategy_id not in self.execution_history:
            return True

        # Get recent executions (last minute)
        recent_executions = [
            exec_time for exec_time in self.execution_history[strategy_id]
            if (datetime.now() - exec_time).total_seconds() < 60
        ]

        return len(recent_executions) < self.max_executions_per_minute

    def record_execution(self, strategy_id: str):
        """Record execution timestamp"""
        if strategy_id not in self.execution_history:
            self.execution_history[strategy_id] = []

        self.execution_history[strategy_id].append(datetime.now())

        # Clean old records
        self.execution_history[strategy_id] = [
            exec_time for exec_time in self.execution_history[strategy_id]
            if (datetime.now() - exec_time).total_seconds() < 3600  # Keep last hour
        ]

    def get_execution_stats(self, strategy_id: str) -> Dict:
        """Get execution statistics for a strategy"""

        if strategy_id not in self.execution_history:
            return {
                "total_executions": 0,
                "executions_per_minute": 0,
                "last_execution": None
            }

        executions = self.execution_history[strategy_id]
        recent_executions = [
            exec_time for exec_time in executions
            if (datetime.now() - exec_time).total_seconds() < 60
        ]

        return {
            "total_executions": len(executions),
            "executions_per_minute": len(recent_executions),
            "last_execution": executions[-1] if executions else None
        }

class TransactionCostAnalyzer:
    def __init__(self, base_commission=0.001, slippage_model=None):
        self.base_commission = base_commission
        self.slippage_model = slippage_model
        self.cost_history = {}

    def estimate_transaction_cost(self, order_size: float, 
                                market_state: Dict) -> float:
        """Estimate transaction cost for given order size"""

        # Commission cost
        commission_cost = order_size * self.base_commission

        # Slippage cost
        slippage_cost = self.estimate_slippage(order_size, market_state)

        total_cost = commission_cost + slippage_cost
        return total_cost

    def estimate_slippage(self, order_size: float, market_state: Dict) -> float:
        """Estimate slippage based on market conditions"""

        volatility = market_state.get("volatility", 0.01)
        liquidity_score = market_state.get("liquidity_score", 1000)

        # Base slippage model
        base_slippage = 0.0005  # 5 basis points

        # Adjust for volatility
        volatility_adjustment = volatility * 10

        # Adjust for liquidity
        liquidity_adjustment = max(0.1, 1000 / liquidity_score) if liquidity_score > 0 else 1.0

        # Adjust for order size
        size_adjustment = min(order_size / 1000, 1.0)  # Normalize to 1000 units

        total_slippage = base_slippage * (1 + volatility_adjustment) * liquidity_adjustment * size_adjustment

        return total_slippage

    def analyze_frequency_cost_impact(self, current_frequency: str, 
                                    proposed_frequency: str,
                                    avg_order_size: float,
                                    market_state: Dict) -> Dict:
        """Analyze cost impact of frequency change"""

        # Get execution intervals
        intervals = {"high": 1, "medium": 60, "low": 300}
        current_interval = intervals[current_frequency]
        proposed_interval = intervals[proposed_frequency]

        # Calculate executions per hour
        current_executions_per_hour = 3600 / current_interval
        proposed_executions_per_hour = 3600 / proposed_interval

        # Calculate costs
        current_cost_per_execution = self.estimate_transaction_cost(avg_order_size, market_state)
        proposed_cost_per_execution = self.estimate_transaction_cost(avg_order_size, market_state)

        current_total_cost = current_executions_per_hour * current_cost_per_execution
        proposed_total_cost = proposed_executions_per_hour * proposed_cost_per_execution

        cost_change = proposed_total_cost - current_total_cost
        cost_change_percent = (cost_change / current_total_cost * 100) if current_total_cost > 0 else 0

        return {
            "current_frequency": current_frequency,
            "proposed_frequency": proposed_frequency,
            "current_cost_per_hour": current_total_cost,
            "proposed_cost_per_hour": proposed_total_cost,
            "cost_change": cost_change,
            "cost_change_percent": cost_change_percent,
            "recommendation": "increase" if cost_change_percent < -10 else "decrease" if cost_change_percent > 10 else "maintain"
        }

from fastapi import APIRouter, HTTPException
from typing import Dict, List, Optional
from pydantic import BaseModel

router = APIRouter()

class FrequencyConfig(BaseModel):
    mode: str = "adaptive"  # "fixed", "event_driven", "adaptive"
    base_frequency: str = "medium"
    max_executions_per_minute: int = 60
    volatility_thresholds: List[float] = [0.01, 0.02, 0.03]

class FrequencyStatus(BaseModel):
    current_frequency: str
    market_regime: str
    execution_stats: Dict
    cost_analysis: Optional[Dict] = None

@router.get("/frequency/status/{strategy_id}")
async def get_frequency_status(strategy_id: str):
    """Get current frequency status for a strategy"""
    try:
        status = {
            "current_frequency": frequency_decider.current_frequency,
            "market_regime": market_state_monitor.get_latest_regime(),
            "execution_stats": execution_rate_limiter.get_execution_stats(strategy_id),
            "last_frequency_change": frequency_decider.get_last_change()
        }
        return status
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@router.post("/frequency/configure")
async def configure_frequency(config: FrequencyConfig):
    """Configure frequency control parameters"""
    try:
        # Update configuration
        frequency_decider.update_config(config.dict())
        schedule_controller.mode = config.mode
        execution_rate_limiter.max_executions_per_minute = config.max_executions_per_minute

        return {
            "status": "configured",
            "config": config.dict()
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@router.get("/frequency/history/{strategy_id}")
async def get_frequency_history(strategy_id: str, limit: int = 100):
    """Get frequency change history for a strategy"""
    try:
        history = frequency_decider.frequency_history[-limit:]
        return {
            "strategy_id": strategy_id,
            "history": history,
            "total_changes": len(frequency_decider.frequency_history)
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@router.get("/frequency/analysis/{strategy_id}")
async def get_frequency_analysis(strategy_id: str):
    """Get frequency analysis and recommendations"""
    try:
        market_state = market_state_monitor.get_market_state(strategy_id)
        current_frequency = frequency_decider.current_frequency

        # Analyze different frequency options
        analysis = {}
        for freq in ["low", "medium", "high"]:
            if freq != current_frequency:
                cost_impact = transaction_cost_analyzer.analyze_frequency_cost_impact(
                    current_frequency, freq, 1000, market_state
                )
                analysis[freq] = cost_impact

        return {
            "strategy_id": strategy_id,
            "current_frequency": current_frequency,
            "market_state": market_state,
            "frequency_analysis": analysis
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@router.post("/frequency/override/{strategy_id}")
async def override_frequency(strategy_id: str, frequency: str):
    """Override frequency for a specific strategy"""
    try:
        if frequency not in ["low", "medium", "high"]:
            raise HTTPException(status_code=400, detail="Invalid frequency")

        # Override frequency
        frequency_decider.current_frequency = frequency
        schedule_controller.current_frequency = frequency

        return {
            "status": "overridden",
            "strategy_id": strategy_id,
            "frequency": frequency
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

// FrequencyControlView.tsx
interface FrequencyStatus {
  currentFrequency: 'low' | 'medium' | 'high';
  marketRegime: string;
  executionStats: {
    totalExecutions: number;
    executionsPerMinute: number;
    lastExecution?: string;
  };
  lastFrequencyChange?: {
    timestamp: string;
    oldFrequency: string;
    newFrequency: string;
    reason: string;
  };
}

interface FrequencyAnalysis {
  currentFrequency: string;
  marketState: Record<string, any>;
  frequencyAnalysis: Record<string, any>;
}

const FrequencyControlView: React.FC = () => {
  const [frequencyStatus, setFrequencyStatus] = useState<FrequencyStatus | null>(null);
  const [frequencyAnalysis, setFrequencyAnalysis] = useState<FrequencyAnalysis | null>(null);
  const [frequencyHistory, setFrequencyHistory] = useState<any[]>([]);

  const getFrequencyStatus = async (strategyId: string) => {
    const response = await api.get(`/frequency/status/${strategyId}`);
    setFrequencyStatus(response.data);
  };

  const getFrequencyAnalysis = async (strategyId: string) => {
    const response = await api.get(`/frequency/analysis/${strategyId}`);
    setFrequencyAnalysis(response.data);
  };

  const overrideFrequency = async (strategyId: string, frequency: string) => {
    await api.post(`/frequency/override/${strategyId}`, { frequency });
    // Refresh status
    await getFrequencyStatus(strategyId);
  };

  return (
    <div className="frequency-control-dashboard">
      {/* Current Frequency Status */}
      <FrequencyStatusPanel 
        status={frequencyStatus}
        onOverride={overrideFrequency}
      />

      {/* Market State Monitor */}
      <MarketStatePanel 
        marketState={frequencyAnalysis?.marketState}
      />

      {/* Frequency Analysis */}
      <FrequencyAnalysisPanel 
        analysis={frequencyAnalysis?.frequencyAnalysis}
      />

      {/* Frequency History */}
      <FrequencyHistoryPanel 
        history={frequencyHistory}
      />

      {/* Execution Statistics */}
      <ExecutionStatsPanel 
        stats={frequencyStatus?.executionStats}
      />
    </div>
  );
};