Research Pipeline: From Strategy to Deployment

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


@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 numpy as np
import pandas as pd


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 numpy as np
import pandas as pd


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(),
}

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Dict, List

import numpy as np
import pandas as pd


@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 hashlib
from datetime import datetime
from pathlib import Path
from typing import Any

import numpy as np
import pandas as pd


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)

from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import Dict, List

import numpy as np
import pandas as pd


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()

from dataclasses import dataclass
from typing import Tuple

import pandas as pd


@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,
)

# Run validation checks
results_df = checklist.run_all_checks()
is_ready = checklist.is_ready_for_production()

from abc import ABC, abstractmethod
from datetime import datetime
from enum import Enum
from typing import Dict

import numpy as np
import pandas as pd


class StrategyState(Enum):
    STOPPED = "stopped"
    RUNNING = "running"
    PAUSED = "paused"
    ERROR = "error"


class TradingStrategy(ABC):
    """Base class for production trading strategies."""

    def __init__(self, strategy_id: str, config: dict):
        self.strategy_id = strategy_id
        self.config = config
        self.state = StrategyState.STOPPED
        self.last_run_time = None
        self.last_run_status = None
        self.error_count = 0
        self.max_errors_before_halt = 3

    @abstractmethod
    def generate_signals(self) -> pd.DataFrame:
        """Generate trading signals. Override in subclass."""
        pass

    @abstractmethod
    def calculate_target_positions(
        self, signals: pd.DataFrame
    ) -> Dict[str, float]:
        """Convert signals to target positions. Override in subclass."""
        pass

    def run(self) -> Dict:
        """Execute one iteration of the strategy."""
        if self.state == StrategyState.PAUSED:
            return {"status": "paused", "message": "Strategy is paused"}

        if (
            self.state == StrategyState.ERROR
            and self.error_count >= self.max_errors_before_halt
        ):
            return {
                "status": "halted",
                "message": "Too many errors, manual intervention required",
            }

        self.state = StrategyState.RUNNING
        self.last_run_time = datetime.now()

        try:
            # Generate signals
            signals = self.generate_signals()

            # Calculate target positions
            targets = self.calculate_target_positions(signals)

            self.last_run_status = "success"
            self.error_count = 0

            return {
                "status": "success",
                "signals": signals,
                "target_positions": targets,
                "timestamp": self.last_run_time.isoformat(),
            }

        except Exception as e:
            self.error_count += 1
            self.last_run_status = "error"

            if self.error_count >= self.max_errors_before_halt:
                self.state = StrategyState.ERROR

            return {
                "status": "error",
                "message": str(e),
                "error_count": self.error_count,
            }

    def pause(self):
        """Pause strategy execution."""
        self.state = StrategyState.PAUSED

    def resume(self):
        """Resume strategy execution."""
        if self.state == StrategyState.PAUSED:
            self.state = StrategyState.STOPPED
            self.error_count = 0

    def get_status(self) -> Dict:
        """Get current strategy status."""
        return {
            "strategy_id": self.strategy_id,
            "state": self.state.value,
            "last_run_time": self.last_run_time.isoformat()
            if self.last_run_time
            else None,
            "last_run_status": self.last_run_status,
            "error_count": self.error_count,
        }


class MomentumStrategy(TradingStrategy):
    """Example momentum strategy implementation."""

    def __init__(self, strategy_id: str, config: dict):
        super().__init__(strategy_id, config)
        self.lookback = config.get("lookback_days", 252)
        self.num_positions = config.get("num_positions", 20)

    def generate_signals(self) -> pd.DataFrame:
        """Generate momentum signals (simplified for demonstration)."""
        # In production, fetch live data from data feed
        np.random.seed(42)
        symbols = [
            "AAPL",
            "GOOGL",
            "MSFT",
            "AMZN",
            "NVDA",
            "META",
            "TSLA",
            "JPM",
            "V",
            "JNJ",
        ]

        signals = pd.DataFrame(
            {
                "symbol": symbols,
                "momentum_score": np.random.randn(len(symbols)),
                "volatility": np.random.uniform(0.15, 0.45, len(symbols)),
            }
        )

        signals["rank"] = signals["momentum_score"].rank(ascending=False)
        return signals

    def calculate_target_positions(
        self, signals: pd.DataFrame
    ) -> Dict[str, float]:
        """Convert signals to target dollar positions."""
        portfolio_value = self.config.get("portfolio_value", 1_000_000)

        # Long top N, short bottom N
        n = min(self.num_positions // 2, len(signals) // 2)

        long_stocks = signals.nsmallest(n, "rank")["symbol"].tolist()
        short_stocks = signals.nlargest(n, "rank")["symbol"].tolist()

        position_size = portfolio_value / (2 * n)

        targets = {}
        for symbol in long_stocks:
            targets[symbol] = position_size
        for symbol in short_stocks:
            targets[symbol] = -position_size

        return targets


# Create and run strategy
momentum_config = {
    "lookback_days": 252,
    "num_positions": 10,
    "portfolio_value": 1_000_000,
}

strategy = MomentumStrategy("MOM_001", momentum_config)

# Execute strategy
result = strategy.run()
status = strategy.get_status()

from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import Dict, List

import numpy as np


class AlertLevel(Enum):
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"


@dataclass
class Alert:
    """Represents a monitoring alert."""

    alert_id: str
    level: AlertLevel
    metric_name: str
    current_value: float
    threshold: float
    message: str
    timestamp: datetime = None
    acknowledged: bool = False

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


class StrategyMonitor:
    """
    Monitor strategy performance and generate alerts.
    """

    def __init__(self, strategy_id: str):
        self.strategy_id = strategy_id
        self.metrics_history: Dict[str, List] = defaultdict(list)
        self.alerts: List[Alert] = []
        self.alert_counter = 0

        # Define alert thresholds
        self.thresholds = {
            "daily_pnl_warning": -10_000,
            "daily_pnl_critical": -25_000,
            "drawdown_warning": -0.05,
            "drawdown_critical": -0.10,
            "position_concentration_warning": 0.15,
            "sharpe_degradation_warning": 0.5,  # Min acceptable Sharpe
        }

    def record_metric(
        self, metric_name: str, value: float, timestamp: datetime = None
    ):
        """Record a metric value."""
        if timestamp is None:
            timestamp = datetime.now()

        self.metrics_history[metric_name].append(
            {"value": value, "timestamp": timestamp}
        )

        # Check for alert conditions
        self._check_alerts(metric_name, value)

    def _check_alerts(self, metric_name: str, value: float):
        """Check if current value triggers any alerts."""

        if metric_name == "daily_pnl":
            if value < self.thresholds["daily_pnl_critical"]:
                self._create_alert(
                    AlertLevel.CRITICAL,
                    metric_name,
                    value,
                    self.thresholds["daily_pnl_critical"],
                    f"Daily P&L ${value:,.0f} below critical threshold",
                )
            elif value < self.thresholds["daily_pnl_warning"]:
                self._create_alert(
                    AlertLevel.WARNING,
                    metric_name,
                    value,
                    self.thresholds["daily_pnl_warning"],
                    f"Daily P&L ${value:,.0f} below warning threshold",
                )

        elif metric_name == "drawdown":
            if value < self.thresholds["drawdown_critical"]:
                self._create_alert(
                    AlertLevel.CRITICAL,
                    metric_name,
                    value,
                    self.thresholds["drawdown_critical"],
                    f"Drawdown {value:.1%} exceeds critical threshold",
                )
            elif value < self.thresholds["drawdown_warning"]:
                self._create_alert(
                    AlertLevel.WARNING,
                    metric_name,
                    value,
                    self.thresholds["drawdown_warning"],
                    f"Drawdown {value:.1%} exceeds warning threshold",
                )

        elif metric_name == "rolling_sharpe":
            if value < self.thresholds["sharpe_degradation_warning"]:
                self._create_alert(
                    AlertLevel.WARNING,
                    metric_name,
                    value,
                    self.thresholds["sharpe_degradation_warning"],
                    f"Rolling Sharpe {value:.2f} below minimum threshold",
                )

    def _create_alert(
        self,
        level: AlertLevel,
        metric_name: str,
        value: float,
        threshold: float,
        message: str,
    ):
        """Create and store an alert."""
        self.alert_counter += 1
        alert = Alert(
            alert_id=f"ALERT_{self.strategy_id}_{self.alert_counter:04d}",
            level=level,
            metric_name=metric_name,
            current_value=value,
            threshold=threshold,
            message=message,
        )
        self.alerts.append(alert)

    def get_active_alerts(self) -> List[Alert]:
        """Get unacknowledged alerts."""
        return [a for a in self.alerts if not a.acknowledged]

    def get_dashboard_data(self) -> Dict:
        """Get data for monitoring dashboard."""
        dashboard = {
            "strategy_id": self.strategy_id,
            "last_updated": datetime.now().isoformat(),
            "active_alerts": len(self.get_active_alerts()),
            "metrics": {},
        }

        for metric_name, history in self.metrics_history.items():
            if history:
                recent = history[-1]
                dashboard["metrics"][metric_name] = {
                    "current_value": recent["value"],
                    "last_updated": recent["timestamp"].isoformat(),
                }

        return dashboard


# Simulate monitoring a strategy
monitor = StrategyMonitor("MOM_001")

# Simulate recording daily metrics over a period
np.random.seed(789)
base_pnl = 5000
base_nav = 1_000_000
cumulative_pnl = 0

for day in range(20):
    # Simulate daily P&L with occasional bad days
    daily_pnl = np.random.normal(
        base_pnl * (1 - day / 100), 15000
    )  # Degrading performance
    cumulative_pnl += daily_pnl
    nav = base_nav + cumulative_pnl

    # Calculate metrics
    drawdown = min(0, cumulative_pnl / base_nav)

    # Estimate rolling Sharpe (simplified)
    returns = [np.random.normal(0.001, 0.02) for _ in range(20)]
    rolling_sharpe = np.mean(returns) / np.std(returns) * np.sqrt(252)

    # Record metrics
    monitor.record_metric("daily_pnl", daily_pnl)
    monitor.record_metric("cumulative_pnl", cumulative_pnl)
    monitor.record_metric("nav", nav)
    monitor.record_metric("drawdown", drawdown)
    monitor.record_metric("rolling_sharpe", rolling_sharpe)