pip install requests faker

import time
import random
import logging
from collections import deque
from typing import Dict, Any, List, Callable

import pandas as pd
from faker import Faker

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

def create_order_monitor_state(alert_threshold: float = 0.8, max_retries: int = 3, base_delay: float = 0.5) -> Dict[str, Any]:
    """
    Initializes the state for the order monitoring system.

    Parameters
    ----------
    alert_threshold : float, optional
        The minimum percentage fill (0.0 to 1.0) to trigger an alert, defaults to 0.8.
    max_retries : int, optional
        Maximum number of retries for simulated external calls, defaults to 3.
    base_delay : float, optional
        Base delay in seconds for exponential backoff, defaults to 0.5.

    Returns
    -------
    Dict[str, Any]
        An initialized state dictionary containing active orders, configuration, and Faker instance.
    """
    state = {
        "active_orders": {},
        "config": {
            "alert_threshold": alert_threshold,
            "max_retries": max_retries,
            "base_delay": base_delay
        },
        "faker": Faker()
    }
    logger.info("Order monitor state initialized.")
    return state

from typing import Dict, Any

def create_order(state: Dict[str, Any], symbol: str, order_type: str, quantity: int, price: float) -> Dict[str, Any]:
    """
    Simulates the creation of a new order and adds it to the active orders in the state.

    Parameters
    ----------
    state : Dict[str, Any]
        The current application state dictionary.
    symbol : str
        The trading symbol for the order (e.g., 'AAPL', 'GOOG').
    order_type : str
        The type of order (e.g., 'LIMIT', 'MARKET').
    quantity : int
        The number of shares/units for the order.
    price : float
        The limit price for the order. For market orders, this might be the current market price or just an indicator.

    Returns
    -------
    Dict[str, Any]
        The updated state dictionary with the new order added.
    """
    order_id = str(state['faker'].uuid4())
    new_order = {
        "order_id": order_id,
        "symbol": symbol,
        "order_type": order_type,
        "initial_quantity": quantity,
        "filled_quantity": 0,
        "price": price,
        "status": "OPEN",
        "created_at": time.time(),
        "last_updated": time.time()
    }
    state["active_orders"][order_id] = new_order
    logger.info(f"Order {order_id} created: {symbol} {quantity} @ {price}")
    return state

from typing import Dict, Any

def simulate_market_data(state: Dict[str, Any], symbol: str, base_price: float, volatility: float = 0.01) -> Dict[str, float]:
    """
    Simulates market data for a given symbol, generating a random price around a base price.

    Parameters
    ----------
    state : Dict[str, Any]
        The current application state dictionary (not modified in this function).
    symbol : str
        The trading symbol for which to simulate market data.
    base_price : float
        The average price around which to generate simulated prices.
    volatility : float, optional
        The magnitude of random fluctuation in price, defaults to 0.01.

    Returns
    -------
    Dict[str, float]
        A dictionary containing the simulated market data, specifically the `last_price`.
    """
    # Add random jitter to simulate market price fluctuations
    price_jitter = random.uniform(-base_price * volatility, base_price * volatility)
    current_price = base_price + price_jitter
    market_data = {
        "symbol": symbol,
        "last_price": round(current_price, 2),
        "timestamp": time.time()
    }
    logger.debug(f"Simulated market data for {symbol}: {market_data['last_price']}")
    return market_data

from typing import Dict, Any

def check_order_fill(state: Dict[str, Any], order_id: str, market_data: Dict[str, float]) -> Dict[str, Any]:
    """
    Checks if an open order has been filled based on provided market data and updates its status.

    Parameters
    ----------
    state : Dict[str, Any]
        The current application state dictionary.
    order_id : str
        The ID of the order to check.
    market_data : Dict[str, float]
        The latest market data containing the `price` for the order's symbol.

    Returns
    -------
    Dict[str, Any]
        The updated state dictionary with the order's status potentially modified.
    """
    if order_id not in state["active_orders"]:
        logger.warning(f"Order {order_id} not found in active orders.")
        return state

    order = state["active_orders"][order_id]
    if order["status"] != "OPEN":
        logger.debug(f"Order {order_id} is not open (status: {order['status']}).")
        return state

    current_market_price = market_data["price"]
    order_filled = False

    if order["order_type"] == "LIMIT":
        # Assuming a BUY limit order: fill if market price <= limit price
        # Assuming a SELL limit order: fill if market price >= limit price
        # For simplicity, we'll assume a BUY order for this example
        if current_market_price <= order["price"]:
            order["filled_quantity"] = order["initial_quantity"]
            order["status"] = "FILLED"
            order["last_updated"] = time.time()
            order_filled = True
            logger.info(f"Limit BUY order {order_id} filled at {current_market_price} (limit was {order['price']}).")
    elif order["order_type"] == "MARKET":
        # Market orders are assumed to be filled immediately at current market price
        order["filled_quantity"] = order["initial_quantity"]
        order["status"] = "FILLED"
        order["price"] = current_market_price # Update price to actual fill price
        order["last_updated"] = time.time()
        order_filled = True
        logger.info(f"Market order {order_id} filled at {current_market_price}.")

    if order_filled:
        logger.info(f"Order {order_id} (Symbol: {order['symbol']}) is now {order['status']}.")

    return state

from typing import Dict, Any

def send_alert(state: Dict[str, Any], order_id: str, message: str) -> bool:
    """
    Simulates sending an alert for a filled order.

    Parameters
    ----------
    state : Dict[str, Any]
        The current application state dictionary (not modified here).
    order_id : str
        The ID of the order for which the alert is being sent.
    message : str
        The custom message to include in the alert.

    Returns
    -------
    bool
        True if the alert was 'sent' successfully, False otherwise (simulated).
    """
    try:
        # Simulate a potential failure with a small chance
        if random.random() < 0.05: # 5% chance of simulated failure
            raise IOError("Simulated network error sending alert.")

        logger.info(f"ALERT for Order {order_id}: {message}")
        # In a real system, integrate with an actual alerting service (e.g., Twilio, SendGrid)
        return True
    except Exception as e:
        logger.error(f"Failed to send alert for Order {order_id}: {e}")
        return False

import time
import random
from typing import Dict, Any, Callable

def execute_with_exponential_backoff(state: Dict[str, Any], func: Callable, *args, **kwargs) -> Any:
    """
    Executes a function with exponential backoff and retries.

    Parameters
    ----------
    state : Dict[str, Any]
        The current application state dictionary, containing max_retries and base_delay in its config.
    func : Callable
        The function to execute.
    *args
        Positional arguments to pass to the function.
    **kwargs
        Keyword arguments to pass to the function.

    Returns
    -------
    Any
        The result of the executed function if successful.

    Raises
    -------
    Exception
        If the function fails after all retries.
    """
    max_retries = state['config'].get('max_retries', 3)
    base_delay = state['config'].get('base_delay', 0.5)

    for attempt in range(max_retries):
        try:
            return func(state, *args, **kwargs)
        except Exception as e:
            delay = base_delay * (2 ** attempt) + random.uniform(0, 0.1) # Add jitter
            logger.warning(f"Attempt {attempt + 1}/{max_retries} failed for {func.__name__}. Retrying in {delay:.2f} seconds. Error: {e}")
            time.sleep(delay)
    logger.error(f"Function {func.__name__} failed after {max_retries} attempts.")
    raise Exception(f"Max retries exceeded for {func.__name__}.")

from typing import Dict, Any

def update_order_status_and_alert(state: Dict[str, Any], order_id: str) -> Dict[str, Any]:
    """
    Updates an order's status and sends an alert if it meets the criteria.

    Parameters
    ----------
    state : Dict[str, Any]
        The current application state dictionary.
    order_id : str
        The ID of the order to update and potentially alert for.

    Returns
    -------
    Dict[str, Any]
        The updated state dictionary.
    """
    order = state["active_orders"].get(order_id)
    if not order:
        logger.warning(f"Order {order_id} not found for status update and alert.")
        return state

    initial_qty = order['initial_quantity']
    filled_qty = order['filled_quantity']
    current_status = order['status']

    if initial_qty > 0:
        fill_percentage = filled_qty / initial_qty
    else:
        fill_percentage = 0.0

    alert_threshold = state['config']['alert_threshold']

    if current_status == "FILLED" and not order.get('alert_sent_filled'):
        alert_message = f"Order {order_id} for {order['symbol']} is FULLY FILLED! Total filled: {filled_qty} at avg price {order['price']:.2f}."
        try:
            execute_with_exponential_backoff(state, send_alert, order_id, alert_message)
            order['alert_sent_filled'] = True
            logger.info(f"Fully filled alert sent for order {order_id}.")
        except Exception as e:
            logger.error(f"Failed to send fully filled alert for order {order_id}: {e}")

    elif fill_percentage >= alert_threshold and not order.get('alert_sent_partial'):
        alert_message = f"Order {order_id} for {order['symbol']} is {fill_percentage:.2%} filled! Filled: {filled_qty}/{initial_qty}."
        try:
            execute_with_exponential_backoff(state, send_alert, order_id, alert_message)
            order['alert_sent_partial'] = True
            logger.info(f"Partial fill alert sent for order {order_id}.")
        except Exception as e:
            logger.error(f"Failed to send partial fill alert for order {order_id}: {e}")

    state["active_orders"][order_id] = order
    return state

import logging

# Configure logging if not already configured (ensures logger is available if this cell is run out of order)
if not logging.getLogger(__name__).handlers:
    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Initialize the state
state = create_order_monitor_state(alert_threshold=0.75, max_retries=3, base_delay=0.2)

# Create some sample orders for cryptocurrencies
state = create_order(state, symbol='BTC', order_type='LIMIT', quantity=0.1, price=65000.00) # Example BTC order
state = create_order(state, symbol='ETH', order_type='MARKET', quantity=1.5, price=3000.00) # Example ETH order
state = create_order(state, symbol='XRP', order_type='LIMIT', quantity=1000, price=0.52) # Example XRP order

logger.info("Initial active orders:")
for order_id, order_details in state['active_orders'].items():
    print(f"  Order ID: {order_id}, Symbol: {order_details['symbol']}, Type: {order_details['order_type']}, Qty: {order_details['initial_quantity']}, Price: {order_details['price']}, Status: {order_details['status']}")

import matplotlib.pyplot as plt
import seaborn as sns

# Data collection for visualization
market_price_history = []
order_status_history = {order_id: [] for order_id in state['active_orders'].keys()}

# Base prices for simulation for cryptocurrencies
base_prices = {'BTC': 65500.00, 'ETH': 3050.00, 'XRP': 0.53}

num_iterations = 50
logger.info(f"Starting {num_iterations} market simulation iterations...")

for i in range(num_iterations):
    current_time = time.time()
    logger.info(f"--- Iteration {i+1}/{num_iterations} ---")

    for symbol, base_price in base_prices.items():
        market_data = simulate_market_data(state, symbol, base_price, volatility=0.005) # Increased volatility for crypto
        market_price_history.append({
            'time': current_time,
            'symbol': symbol,
            'price': market_data['last_price']
        })

    for order_id in list(state['active_orders'].keys()): # Iterate over a copy to allow modification
        order = state['active_orders'][order_id]
        if order['status'] == 'FILLED':
            continue # Skip filled orders

        # Get the latest market data for the order's symbol
        # In this simulation, we'll just pass the latest market_data generated for the current iteration and symbol
        # For a real system, you'd fetch the most recent data for that specific symbol
        current_market_data_for_order = next((md for md in market_price_history if md['symbol'] == order['symbol'] and md['time'] == current_time), None)

        if current_market_data_for_order:
            state = check_order_fill(state, order_id, current_market_data_for_order)
            state = update_order_status_and_alert(state, order_id)

        # Record order status for visualization
        order_status_history[order_id].append({
            'time': current_time,
            'status': state['active_orders'][order_id]['status'],
            'filled_quantity': state['active_orders'][order_id]['filled_quantity'],
            'total_quantity': state['active_orders'][order_id]['initial_quantity']
        })

    time.sleep(0.1) # Simulate time passing

logger.info("Market simulation complete.")

import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns

# Convert histories to DataFrames for easier plotting
market_df = pd.DataFrame(market_price_history)

order_dfs = []
for order_id, history in order_status_history.items():
    df = pd.DataFrame(history)
    if not df.empty:
        df['order_id'] = order_id
        df['symbol'] = state['active_orders'][order_id]['symbol']
        df['fill_percentage'] = df['filled_quantity'] / df['total_quantity']
        order_dfs.append(df)

if order_dfs:
    all_orders_df = pd.concat(order_dfs, ignore_index=True)
else:
    all_orders_df = pd.DataFrame()


# Set plot style
sns.set_style("whitegrid")

fig, axes = plt.subplots(2, 1, figsize=(14, 12), sharex=True)

# Plot 1: Market Prices over Time
sns.lineplot(data=market_df, x='time', y='price', hue='symbol', ax=axes[0])
axes[0].set_title('Simulated Market Prices Over Time')
axes[0].set_ylabel('Price ($)')
axes[0].legend(title='Symbol')
axes[0].grid(True)

# Plot 2: Order Fill Progress
if not all_orders_df.empty:
    sns.lineplot(data=all_orders_df, x='time', y='fill_percentage', hue='symbol', style='order_id', ax=axes[1])
    axes[1].axhline(y=state['config']['alert_threshold'], color='r', linestyle='--', label='Alert Threshold')
    axes[1].set_title('Order Fill Percentage Over Time')
    axes[1].set_ylabel('Fill Percentage')
    axes[1].set_xlabel('Time')
    axes[1].set_ylim(0, 1.1) # Ensure y-axis covers 0-100% and a bit more
    axes[1].legend(title='Order')
    axes[1].grid(True)
else:
    axes[1].text(0.5, 0.5, 'No order fill data to display', horizontalalignment='center', verticalalignment='center', transform=axes[1].transAxes)

plt.tight_layout()
plt.show()

import pandas as pd

# Prepare data for summary DataFrame
summary_data = []
for order_id, order in state['active_orders'].items():
    summary_data.append({
        "Order ID": order_id,
        "Symbol": order['symbol'],
        "Order Type": order['order_type'],
        "Initial Quantity": order['initial_quantity'],
        "Filled Quantity": order['filled_quantity'],
        "Fill Price": f"{order['price']:.2f}" if order['status'] == 'FILLED' else 'N/A',
        "Status": order['status'],
        "Created At": pd.to_datetime(order['created_at'], unit='s'),
        "Last Updated": pd.to_datetime(order['last_updated'], unit='s')
    })

summary_df = pd.DataFrame(summary_data)

print("\n--- Order Summary ---")
display(summary_df)

# Optional: further analysis like count of filled/open orders
filled_count = summary_df[summary_df['Status'] == 'FILLED'].shape[0]
open_count = summary_df[summary_df['Status'] == 'OPEN'].shape[0]

print(f"\nTotal Orders: {len(summary_df)}")
print(f"Filled Orders: {filled_count}")
print(f"Open Orders: {open_count}")