Research Pipeline & Deployment: Strategy Lifecycle Guide

from dataclasses import dataclass
from typing import Optional
from datetime import datetime


@dataclass
class ResearchHypothesis:
    """Document a trading hypothesis with all relevant details."""

    name: str
    description: str
    asset_universe: str
    signal_definition: str
    expected_effect: str
    holding_period: str
    economic_rationale: str
    conditions: Optional[str] = None
    created_date: str = None

    def __post_init__(self):
        if self.created_date is None:
            self.created_date = datetime.now().strftime("%Y-%m-%d")

    def to_dict(self) -> dict:
        return {
            "name": self.name,
            "description": self.description,
            "asset_universe": self.asset_universe,
            "signal_definition": self.signal_definition,
            "expected_effect": self.expected_effect,
            "holding_period": self.holding_period,
            "economic_rationale": self.economic_rationale,
            "conditions": self.conditions,
            "created_date": self.created_date,
        }


# Example hypothesis documentation
momentum_hypothesis = ResearchHypothesis(
    name="Cross-Sectional Momentum",
    description="Past winners continue to outperform past losers",
    asset_universe="S&P 500 constituents",
    signal_definition="12-month return excluding most recent month (12-1 momentum)",
    expected_effect="Top decile outperforms bottom decile by 0.4-0.8% per month",
    holding_period="1 month with monthly rebalancing",
    economic_rationale="Behavioral underreaction to news and positive feedback trading",
    conditions="Effect stronger in high-attention periods and for stocks with recent news",
)

import pandas as pd
import numpy as np


class DataQualityChecker:
    """Check common data quality issues in financial datasets."""

    def __init__(self, prices: pd.DataFrame, returns: pd.DataFrame = None):
        self.prices = prices
        self.returns = returns if returns is not None else prices.pct_change()
        self.issues = []

    def check_missing_data(self, threshold: float = 0.1) -> dict:
        """Check for excessive missing data."""
        missing_pct = self.prices.isnull().sum() / len(self.prices)
        problematic = missing_pct[missing_pct > threshold]

        if len(problematic) > 0:
            self.issues.append(
                f"High missing data: {len(problematic)} columns > {threshold:.0%}"
            )

        return {
            "total_missing_pct": self.prices.isnull().sum().sum()
            / self.prices.size,
            "columns_above_threshold": len(problematic),
            "worst_columns": problematic.nlargest(5).to_dict(),
        }

    def check_return_outliers(self, threshold: float = 0.5) -> dict:
        """Check for unrealistic return spikes (potential data errors)."""
        extreme_returns = (self.returns.abs() > threshold).sum()
        problematic = extreme_returns[extreme_returns > 0]

        if len(problematic) > 0:
            total_extreme = problematic.sum()
            self.issues.append(
                f"Extreme returns (>{threshold:.0%}): {total_extreme} observations"
            )

        return {
            "columns_with_extremes": len(problematic),
            "total_extreme_observations": problematic.sum(),
            "max_return": self.returns.max().max(),
            "min_return": self.returns.min().min(),
        }

    def check_stale_prices(self, consecutive_days: int = 5) -> dict:
        """Check for suspiciously unchanged prices."""
        unchanged = (self.returns == 0).astype(int)

        # Count consecutive zeros using rolling sum
        stale_count = 0
        for col in unchanged.columns:
            rolling_zeros = unchanged[col].rolling(consecutive_days).sum()
            stale_count += (rolling_zeros == consecutive_days).sum()

        if stale_count > 0:
            self.issues.append(
                f"Stale prices: {stale_count} instances of {consecutive_days}+ unchanged days"
            )

        return {
            "stale_instances": stale_count,
            "consecutive_days_checked": consecutive_days,
        }

    def generate_report(self) -> pd.DataFrame:
        """Run all checks and generate summary report."""
        missing = self.check_missing_data()
        outliers = self.check_return_outliers()
        stale = self.check_stale_prices()

        report_data = {
            "Check": ["Missing Data", "Return Outliers", "Stale Prices"],
            "Status": [
                "PASS" if missing["columns_above_threshold"] == 0 else "WARN",
                "PASS"
                if outliers["total_extreme_observations"] == 0
                else "WARN",
                "PASS" if stale["stale_instances"] == 0 else "WARN",
            ],
            "Details": [
                f"{missing['total_missing_pct']:.2%} total missing",
                f"{outliers['total_extreme_observations']} extreme returns",
                f"{stale['stale_instances']} stale instances",
            ],
        }

        return pd.DataFrame(report_data)


# Generate sample data with intentional quality issues
np.random.seed(42)
n_days = 252
n_stocks = 10

dates = pd.date_range("2023-01-01", periods=n_days, freq="B")
stock_names = [f"STOCK_{i}" for i in range(n_stocks)]

# Create price data with some issues
prices_data = np.random.randn(n_days, n_stocks).cumsum(axis=0) + 100
prices_data = np.exp(prices_data / 100) * 100  # Convert to realistic prices

# Inject some data quality issues
prices_data[50:55, 3] = prices_data[49, 3]  # Stale prices
prices_data[100, 5] = prices_data[99, 5] * 2.5  # Outlier spike
prices_data[150:160, 7] = np.nan  # Missing data

sample_prices = pd.DataFrame(prices_data, index=dates, columns=stock_names)

## Run data quality checks
checker = DataQualityChecker(sample_prices)
quality_report = checker.generate_report()

import pandas as pd
import numpy as np


class SignalGenerator:
    """Generate trading signals from price data."""

    def __init__(self, prices: pd.DataFrame):
        self.prices = prices.copy()
        self.returns = prices.pct_change()
        self.signals = pd.DataFrame(index=prices.index)

    def momentum(self, lookback: int = 252, skip: int = 21) -> pd.DataFrame:
        """
        Calculate momentum signal (past returns excluding recent period).

        Args:
            lookback: Total lookback period in trading days
            skip: Recent days to exclude (avoids short-term reversal)
        """
        # Calculate returns over lookback period, excluding most recent 'skip' days
        total_return = self.prices.pct_change(lookback)
        recent_return = self.prices.pct_change(skip)

        # Momentum = total return - recent return (approximately)
        momentum_signal = (1 + total_return) / (1 + recent_return) - 1

        self.signals["momentum"] = momentum_signal.mean(
            axis=1
        )  # Portfolio average
        return momentum_signal

    def mean_reversion(self, lookback: int = 21) -> pd.DataFrame:
        """
        Calculate mean reversion signal (deviation from moving average).
        """
        ma = self.prices.rolling(lookback).mean()
        deviation = (self.prices - ma) / ma

        # Mean reversion: buy when price is below MA (negative signal = buy)
        mr_signal = -deviation

        self.signals["mean_reversion"] = mr_signal.mean(axis=1)
        return mr_signal

    def volatility_adjusted_momentum(
        self, lookback: int = 252, vol_lookback: int = 63
    ) -> pd.DataFrame:
        """
        Momentum signal normalized by volatility.
        """
        momentum = self.prices.pct_change(lookback)
        volatility = self.returns.rolling(vol_lookback).std() * np.sqrt(252)

        # Normalize momentum by volatility
        vol_adj_mom = momentum / volatility

        self.signals["vol_adj_momentum"] = vol_adj_mom.mean(axis=1)
        return vol_adj_mom

    def cross_sectional_rank(self, raw_signal: pd.DataFrame) -> pd.DataFrame:
        """
        Convert raw signal to cross-sectional percentile rank.
        Ranks are more robust to outliers.
        """
        return raw_signal.rank(axis=1, pct=True)


# Generate signals from our sample data
signal_gen = SignalGenerator(sample_prices.dropna(axis=1))

momentum_raw = signal_gen.momentum(lookback=126, skip=21)
momentum_ranked = signal_gen.cross_sectional_rank(momentum_raw)

mean_rev_raw = signal_gen.mean_reversion(lookback=21)
mean_rev_ranked = signal_gen.cross_sectional_rank(mean_rev_raw)

# Calculate statistics for display
mom_stats = {
    "mean": momentum_ranked.stack().mean(),
    "std": momentum_ranked.stack().std(),
    "min": momentum_ranked.stack().min(),
    "max": momentum_ranked.stack().max(),
}

mr_stats = {
    "mean": mean_rev_ranked.stack().mean(),
    "std": mean_rev_ranked.stack().std(),
    "min": mean_rev_ranked.stack().min(),
    "max": mean_rev_ranked.stack().max(),
}

import pandas as pd
import numpy as np
from datetime import datetime
from dataclasses import dataclass, field
from typing import List, Dict, Any


@dataclass
class ExperimentResult:
    """Store results from a single experiment run."""

    experiment_id: str
    hypothesis_name: str
    parameters: Dict[str, Any]
    metrics: Dict[str, float]
    data_period: str
    notes: str = ""
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())

    def summary(self) -> str:
        return (
            f"Experiment {self.experiment_id}: "
            f"Sharpe={self.metrics.get('sharpe_ratio', 'N/A'):.2f}, "
            f"Return={self.metrics.get('annual_return', 'N/A'):.1%}"
        )


class ResearchLog:
    """Track all experiments for a research project."""

    def __init__(self, project_name: str):
        self.project_name = project_name
        self.experiments: List[ExperimentResult] = []
        self.next_id = 1

    def log_experiment(
        self,
        hypothesis_name: str,
        parameters: dict,
        metrics: dict,
        data_period: str,
        notes: str = "",
    ) -> ExperimentResult:
        """Log a new experiment result."""
        exp = ExperimentResult(
            experiment_id=f"EXP_{self.next_id:04d}",
            hypothesis_name=hypothesis_name,
            parameters=parameters,
            metrics=metrics,
            data_period=data_period,
            notes=notes,
        )
        self.experiments.append(exp)
        self.next_id += 1
        return exp

    def get_best_experiments(
        self, metric: str = "sharpe_ratio", n: int = 5
    ) -> List[ExperimentResult]:
        """Get top N experiments by specified metric."""
        valid_exps = [e for e in self.experiments if metric in e.metrics]
        return sorted(
            valid_exps, key=lambda x: x.metrics[metric], reverse=True
        )[:n]

    def to_dataframe(self) -> pd.DataFrame:
        """Convert experiment log to DataFrame for analysis."""
        records = []
        for exp in self.experiments:
            record = {
                "experiment_id": exp.experiment_id,
                "hypothesis": exp.hypothesis_name,
                "data_period": exp.data_period,
                "timestamp": exp.timestamp,
                **exp.parameters,
                **exp.metrics,
            }
            records.append(record)
        return pd.DataFrame(records)


# Simulate a research session with multiple parameter combinations
research_log = ResearchLog("Momentum Strategy Research")

# Test different parameter combinations
param_grid = [
    {"lookback": 126, "skip": 21, "holding_period": 21},
    {"lookback": 252, "skip": 21, "holding_period": 21},
    {"lookback": 126, "skip": 5, "holding_period": 21},
    {"lookback": 252, "skip": 21, "holding_period": 63},
]

np.random.seed(123)
for params in param_grid:
    # Simulate backtest results (in practice, run actual backtest)
    metrics = {
        "sharpe_ratio": np.random.uniform(0.3, 1.5),
        "annual_return": np.random.uniform(0.02, 0.15),
        "max_drawdown": np.random.uniform(-0.25, -0.05),
        "win_rate": np.random.uniform(0.48, 0.55),
    }

    research_log.log_experiment(
        hypothesis_name="Cross-Sectional Momentum",
        parameters=params,
        metrics=metrics,
        data_period="2018-01-01 to 2022-12-31",
    )

## Review experiment results
exp_df = research_log.to_dataframe()
best_experiments = research_log.get_best_experiments("sharpe_ratio", n=3)

# Recommended repository structure for quant research
repository_structure = """
quant-research/
├── README.md
├── requirements.txt
├── setup.py
├── .gitignore
│
├── config/
│   ├── strategy_params.yaml
│   └── data_sources.yaml
│
├── data/
│   ├── raw/              # Original data (often gitignored)
│   ├── processed/        # Cleaned data
│   └── features/         # Computed features
│
├── src/
│   ├── __init__.py
│   ├── data/
│   │   ├── loaders.py
│   │   └── cleaners.py
│   ├── features/
│   │   ├── signals.py
│   │   └── transforms.py
│   ├── models/
│   │   ├── portfolio.py
│   │   └── risk.py
│   ├── backtest/
│   │   ├── engine.py
│   │   └── metrics.py
│   └── execution/
│       ├── paper_trading.py
│       └── production.py
│
├── notebooks/
│   ├── 01_data_exploration.ipynb
│   ├── 02_signal_research.ipynb
│   └── 03_backtest_analysis.ipynb
│
├── experiments/
│   ├── configs/          # Experiment configurations
│   └── results/          # Experiment outputs
│
├── tests/
│   ├── test_signals.py
│   └── test_backtest.py
│
└── scripts/
    ├── run_backtest.py
    └── deploy_strategy.py
"""

# Key .gitignore entries for quant research
gitignore_content = """
# Data files (often too large for git)
data/raw/
*.csv
*.parquet
*.h5

# Credentials and secrets
.env
credentials/
*_secret*

# Experiment outputs (track configs, not results)
experiments/results/
*.pkl

# Python
__pycache__/
*.pyc
.venv/
*.egg-info/

# Jupyter
.ipynb_checkpoints/

# IDE
.vscode/
.idea/
"""

import pandas as pd
import numpy as np
from datetime import datetime
from typing import Any
import hashlib
from pathlib import Path


class SimpleExperimentTracker:
    """
    Lightweight experiment tracking inspired by MLflow.
    In production, use MLflow, Weights & Biases, or similar tools.
    """

    def __init__(self, experiment_dir: str = "experiments"):
        self.experiment_dir = Path(experiment_dir)
        self.experiment_dir.mkdir(exist_ok=True)
        self.current_run = None
        self.runs = []

    def start_run(self, run_name: str = None) -> str:
        """Start a new experiment run."""
        run_id = hashlib.md5(
            f"{datetime.now().isoformat()}{run_name}".encode()
        ).hexdigest()[:8]

        self.current_run = {
            "run_id": run_id,
            "run_name": run_name or f"run_{run_id}",
            "start_time": datetime.now().isoformat(),
            "parameters": {},
            "metrics": {},
            "tags": {},
            "artifacts": [],
        }
        return run_id

    def log_param(self, key: str, value: Any):
        """Log a parameter for the current run."""
        if self.current_run is None:
            raise RuntimeError("No active run. Call start_run() first.")
        self.current_run["parameters"][key] = value

    def log_params(self, params: dict):
        """Log multiple parameters."""
        for key, value in params.items():
            self.log_param(key, value)

    def log_metric(self, key: str, value: float, step: int = None):
        """Log a metric for the current run."""
        if self.current_run is None:
            raise RuntimeError("No active run. Call start_run() first.")

        if key not in self.current_run["metrics"]:
            self.current_run["metrics"][key] = []

        self.current_run["metrics"][key].append(
            {
                "value": value,
                "step": step,
                "timestamp": datetime.now().isoformat(),
            }
        )

    def log_metrics(self, metrics: dict, step: int = None):
        """Log multiple metrics."""
        for key, value in metrics.items():
            self.log_metric(key, value, step)

    def set_tag(self, key: str, value: str):
        """Set a tag for the current run."""
        if self.current_run is None:
            raise RuntimeError("No active run. Call start_run() first.")
        self.current_run["tags"][key] = value

    def end_run(self):
        """End the current run and save results."""
        if self.current_run is None:
            return

        self.current_run["end_time"] = datetime.now().isoformat()

        # Simplify metrics to final values for storage
        simplified_metrics = {}
        for key, values in self.current_run["metrics"].items():
            simplified_metrics[key] = values[-1]["value"] if values else None
        self.current_run["final_metrics"] = simplified_metrics

        self.runs.append(self.current_run)
        self.current_run = None

    def get_runs_dataframe(self) -> pd.DataFrame:
        """Get all runs as a DataFrame."""
        records = []
        for run in self.runs:
            record = {
                "run_id": run["run_id"],
                "run_name": run["run_name"],
                "start_time": run["start_time"],
                **run["parameters"],
                **run.get("final_metrics", {}),
            }
            records.append(record)
        return pd.DataFrame(records)


# Demonstrate experiment tracking
tracker = SimpleExperimentTracker()

# Run multiple experiments with tracking
strategies_to_test = [
    {"name": "momentum_fast", "lookback": 63, "holding": 5},
    {"name": "momentum_medium", "lookback": 126, "holding": 21},
    {"name": "momentum_slow", "lookback": 252, "holding": 63},
]

np.random.seed(456)
for strategy in strategies_to_test:
    tracker.start_run(run_name=strategy["name"])

    # Log parameters
    tracker.log_params(
        {
            "lookback_days": strategy["lookback"],
            "holding_period": strategy["holding"],
            "universe": "SP500",
            "start_date": "2018-01-01",
            "end_date": "2022-12-31",
        }
    )

    # Set tags for organization
    tracker.set_tag("strategy_type", "momentum")
    tracker.set_tag("researcher", "quant_team")

    # Simulate running backtest and logging metrics
    for month in range(1, 13):
        monthly_return = np.random.normal(0.005, 0.03)
        tracker.log_metric(
            "cumulative_return", monthly_return * month, step=month
        )

    # Log final metrics
    tracker.log_metrics(
        {
            "sharpe_ratio": np.random.uniform(0.4, 1.2),
            "annual_return": np.random.uniform(0.03, 0.12),
            "max_drawdown": np.random.uniform(-0.20, -0.05),
            "calmar_ratio": np.random.uniform(0.2, 0.8),
        }
    )

    tracker.end_run()

## Review tracked experiments
runs_df = tracker.get_runs_dataframe()
display_cols = [
    "run_name",
    "lookback_days",
    "holding_period",
    "sharpe_ratio",
    "annual_return",
    "max_drawdown",
]

!uv pip install PyYAML
import yaml

# Example strategy configuration
strategy_config = {
    'strategy': {
        'name': 'momentum_strategy_v2',
        'version': '2.1.0',
        'description': 'Cross-sectional momentum with volatility adjustment'
    },
    'universe': {
        'source': 'sp500_constituents',
        'min_price': 5.0,
        'min_volume': 1000000,
        'exclude_sectors': ['Utilities']
    },
    'signal': {
        'type': 'momentum',
        'lookback_days': 252,
        'skip_days': 21,
        'vol_adjustment': True,
        'vol_lookback': 63
    },
    'portfolio': {
        'rebalance_frequency': 'monthly',
        'num_long': 50,
        'num_short': 50,
        'max_position_size': 0.04,
        'target_volatility': 0.10
    },
    'risk': {
        'max_sector_exposure': 0.25,
        'max_single_stock': 0.05,
        'stop_loss': -0.15
    },
    'execution': {
        'order_type': 'VWAP',
        'participation_rate': 0.10,
        'max_spread_bps': 50
    }
}

# Convert to YAML string for display
config_yaml = yaml.dump(strategy_config, default_flow_style=False, sort_keys=False)

import pandas as pd
import numpy as np
from datetime import datetime
from enum import Enum
from typing import Dict, List
from collections import defaultdict
from dataclasses import dataclass


class OrderStatus(Enum):
    PENDING = "pending"
    FILLED = "filled"
    PARTIAL = "partial"
    CANCELLED = "cancelled"
    REJECTED = "rejected"


@dataclass
class Order:
    """Represents a trading order."""

    order_id: str
    symbol: str
    side: str  # 'buy' or 'sell'
    quantity: float
    order_type: str  # 'market', 'limit'
    limit_price: float = None
    status: OrderStatus = OrderStatus.PENDING
    filled_quantity: float = 0
    filled_price: float = None
    timestamp: datetime = None

    def __post_init__(self):
        if self.timestamp is None:
            self.timestamp = datetime.now()


class PaperTradingEngine:
    """
    Simulates order execution and portfolio tracking for paper trading.
    """

    def __init__(self, initial_capital: float = 1_000_000):
        self.initial_capital = initial_capital
        self.cash = initial_capital
        self.positions: Dict[str, float] = defaultdict(float)
        self.orders: Dict[str, Order] = {}
        self.trades: List[Dict] = []
        self.order_counter = 0

        # Simulated market data (in production, connect to live feed)
        self.current_prices: Dict[str, float] = {}

        # Execution simulation parameters
        self.fill_probability = 0.95  # Probability of market order fill
        self.slippage_bps = 5  # Average slippage in basis points

    def update_prices(self, prices: Dict[str, float]):
        """Update current market prices."""
        self.current_prices = prices

    def submit_order(
        self,
        symbol: str,
        side: str,
        quantity: float,
        order_type: str = "market",
        limit_price: float = None,
    ) -> Order:
        """Submit an order to the paper trading system."""
        self.order_counter += 1
        order_id = f"PAPER_{self.order_counter:06d}"

        order = Order(
            order_id=order_id,
            symbol=symbol,
            side=side,
            quantity=quantity,
            order_type=order_type,
            limit_price=limit_price,
        )

        self.orders[order_id] = order

        # Attempt immediate fill for market orders
        if order_type == "market":
            self._attempt_fill(order)

        return order

    def _attempt_fill(self, order: Order):
        """Simulate order execution."""
        if order.symbol not in self.current_prices:
            order.status = OrderStatus.REJECTED
            return

        current_price = self.current_prices[order.symbol]

        # Simulate fill probability
        if np.random.random() > self.fill_probability:
            order.status = OrderStatus.REJECTED
            return

        # Apply slippage
        slippage = current_price * self.slippage_bps / 10000
        if order.side == "buy":
            fill_price = current_price + slippage
        else:
            fill_price = current_price - slippage

        # Check if we have enough cash for buys
        order_value = fill_price * order.quantity
        if order.side == "buy" and order_value > self.cash:
            # Partial fill based on available cash
            affordable_qty = int(self.cash / fill_price)
            if affordable_qty == 0:
                order.status = OrderStatus.REJECTED
                return
            order.filled_quantity = affordable_qty
            order.status = OrderStatus.PARTIAL
        else:
            order.filled_quantity = order.quantity
            order.status = OrderStatus.FILLED

        order.filled_price = fill_price

        # Update positions and cash
        self._update_positions(order)

    def _update_positions(self, order: Order):
        """Update portfolio after a fill."""
        trade_value = order.filled_price * order.filled_quantity

        if order.side == "buy":
            self.positions[order.symbol] += order.filled_quantity
            self.cash -= trade_value
        else:
            self.positions[order.symbol] -= order.filled_quantity
            self.cash += trade_value

        # Record trade
        self.trades.append(
            {
                "order_id": order.order_id,
                "symbol": order.symbol,
                "side": order.side,
                "quantity": order.filled_quantity,
                "price": order.filled_price,
                "value": trade_value,
                "timestamp": datetime.now().isoformat(),
            }
        )

    def get_portfolio_value(self) -> float:
        """Calculate total portfolio value."""
        positions_value = sum(
            qty * self.current_prices.get(symbol, 0)
            for symbol, qty in self.positions.items()
        )
        return self.cash + positions_value

    def get_portfolio_summary(self) -> pd.DataFrame:
        """Get current portfolio positions."""
        records = []
        for symbol, quantity in self.positions.items():
            if quantity != 0:
                price = self.current_prices.get(symbol, 0)
                value = quantity * price
                records.append(
                    {
                        "symbol": symbol,
                        "quantity": quantity,
                        "price": price,
                        "market_value": value,
                        "weight": value / self.get_portfolio_value()
                        if self.get_portfolio_value() > 0
                        else 0,
                    }
                )

        df = pd.DataFrame(records)
        if len(df) > 0:
            df = df.sort_values("market_value", ascending=False)
        return df


# Demonstrate paper trading
np.random.seed(42)
paper_engine = PaperTradingEngine(initial_capital=100_000)

# Simulate market prices
paper_engine.update_prices(
    {
        "AAPL": 175.50,
        "GOOGL": 140.25,
        "MSFT": 378.90,
        "AMZN": 178.50,
        "NVDA": 875.30,
    }
)

# Submit some orders
orders = [
    paper_engine.submit_order("AAPL", "buy", 100),
    paper_engine.submit_order("GOOGL", "buy", 50),
    paper_engine.submit_order("MSFT", "buy", 30),
    paper_engine.submit_order("NVDA", "buy", 10),
]

## Review paper trading results
initial_capital = paper_engine.initial_capital
current_cash = paper_engine.cash
portfolio_value = paper_engine.get_portfolio_value()
pnl = portfolio_value - initial_capital
portfolio_df = paper_engine.get_portfolio_summary()

import pandas as pd
from typing import Tuple
from dataclasses import dataclass


@dataclass
class PaperTradingChecklist:
    """Validation checklist for paper trading readiness."""

    # Minimum requirements before going live
    min_paper_trading_days: int = 30
    max_acceptable_tracking_error: float = 0.02  # 2% vs backtest
    min_fill_rate: float = 0.90
    max_slippage_vs_expected: float = 1.5  # No more than 1.5x expected

    # Tracking metrics
    days_traded: int = 0
    realized_sharpe: float = None
    backtest_sharpe: float = None
    fill_rate: float = None
    avg_slippage_bps: float = None
    expected_slippage_bps: float = None
    system_uptime: float = None
    data_quality_score: float = None

    def check_duration(self) -> Tuple[bool, str]:
        passed = self.days_traded >= self.min_paper_trading_days
        msg = f"Paper traded {self.days_traded}/{self.min_paper_trading_days} days"
        return passed, msg

    def check_tracking_error(self) -> Tuple[bool, str]:
        if self.realized_sharpe is None or self.backtest_sharpe is None:
            return False, "Missing Sharpe ratio data"

        tracking_error = abs(self.realized_sharpe - self.backtest_sharpe) / max(
            self.backtest_sharpe, 0.1
        )
        passed = tracking_error <= self.max_acceptable_tracking_error
        msg = f"Tracking error: {tracking_error:.1%} (max {self.max_acceptable_tracking_error:.0%})"
        return passed, msg

    def check_fill_rate(self) -> Tuple[bool, str]:
        if self.fill_rate is None:
            return False, "Missing fill rate data"

        passed = self.fill_rate >= self.min_fill_rate
        msg = f"Fill rate: {self.fill_rate:.1%} (min {self.min_fill_rate:.0%})"
        return passed, msg

    def check_slippage(self) -> Tuple[bool, str]:
        if self.avg_slippage_bps is None or self.expected_slippage_bps is None:
            return False, "Missing slippage data"

        slippage_ratio = self.avg_slippage_bps / self.expected_slippage_bps
        passed = slippage_ratio <= self.max_slippage_vs_expected
        msg = f"Slippage: {self.avg_slippage_bps:.1f}bps (expected {self.expected_slippage_bps:.1f}bps)"
        return passed, msg

    def run_all_checks(self) -> pd.DataFrame:
        """Run all validation checks and return summary."""
        checks = [
            ("Duration", self.check_duration()),
            ("Tracking Error", self.check_tracking_error()),
            ("Fill Rate", self.check_fill_rate()),
            ("Slippage", self.check_slippage()),
        ]

        results = []
        for check_name, (passed, message) in checks:
            results.append(
                {
                    "Check": check_name,
                    "Status": "PASS" if passed else "FAIL",
                    "Details": message,
                }
            )

        return pd.DataFrame(results)

    def is_ready_for_production(self) -> bool:
        """Determine if strategy passes all checks."""
        checks = [
            self.check_duration()[0],
            self.check_tracking_error()[0],
            self.check_fill_rate()[0],
            self.check_slippage()[0],
        ]
        return all(checks)


# Example checklist evaluation
checklist = PaperTradingChecklist(
    days_traded=45,
    realized_sharpe=1.18,
    backtest_sharpe=1.20,
    fill_rate=0.94,
    avg_slippage_bps=8.5,
    expected_slippage_bps=6.0,
    system_uptime=0.998,
)