Trading System Prefect Architecture¶

┌─────────────────────────────────────────────────────────┐
│                Prefect Server                          │
│                (Port 4200)                             │
├─────────────────────────────────────────────────────────┤
│  Flow Management  │  Task Execution  │  Monitoring     │
│  ├── Deployments  │  ├── Work Pools  │  ├── UI         │
│  ├── Flow Runs    │  ├── Workers     │  ├── Logs       │
│  └── Schedules    │  └── Tasks       │  └── Metrics    │
└─────────────────────────────────────────────────────────┘

@flow(name="fetch_market_data", retries=3, retry_delay_seconds=60)
def fetch_market_data(symbols: List[str], start_time: datetime):
    """Hourly market data collection from Alpaca API"""
    # Fetch data from Alpaca
    # Validate data quality
    # Store in PostgreSQL
    # Cache in Redis
    # Publish data events

@flow(name="validate_data_quality", retries=2)
def validate_data_quality(data: pd.DataFrame):
    """Data validation and quality checks"""
    # Check for missing values
    # Validate price ranges
    # Check volume consistency
    # Flag data quality issues

@flow(name="archive_old_data", schedule=CronSchedule(cron="0 2 * * *"))
def archive_old_data(retention_days: int = 90):
    """Daily data archival and cleanup"""
    # Archive old market data
    # Clean up temporary files
    # Update data retention policies

@flow(
    name="intraday-pairs-trading",
    log_prints=True,
    retries=1,
    retry_delay_seconds=60,
    timeout_seconds=600,
)
async def intraday_pairs_flow(skip_market_check: bool = False) -> dict:
    """
    Hourly intraday pairs trading cycle.
    1. Check market is open (Alpaca clock)
    2. For each active pair: fetch live bars → compute z-score → generate signal → execute
    """
    alpaca = AlpacaClient(is_paper=True)

    if not skip_market_check:
        is_open = await check_market_open_task(alpaca)
        if not is_open:
            return {"status": "MARKET_CLOSED"}

    return await run_pairs_strategy_task(alpaca)

# Dry-run one cycle (market check skipped):
python src/shared/prefect/flows/strategy_engine/pairs_flow.py

# Register scheduled deployment in Prefect UI:
python src/shared/prefect/flows/strategy_engine/pairs_flow.py --deploy

@flow(name="execute_trades", retries=3, retry_delay_seconds=30)
def execute_trades(signals: List[Dict]):
    """Process trading signals and execute trades"""
    # Validate signals
    # Check risk limits
    # Place orders with Alpaca
    # Update order status
    # Log execution details

@flow(name="manage_orders", retries=2)
def manage_orders():
    """Order lifecycle management"""
    # Check pending orders
    # Update order status
    # Handle partial fills
    # Process cancellations

@flow(name="update_positions", retries=1)
def update_positions():
    """Position tracking and P&L calculation"""
    # Fetch current positions
    # Calculate unrealized P&L
    # Update position records
    # Generate position reports

@flow(name="reconcile_trades", schedule=CronSchedule(cron="0 */6 * * *"))
def reconcile_trades():
    """Trade reconciliation with broker"""
    # Fetch trades from Alpaca
    # Compare with local records
    # Resolve discrepancies
    # Update trade status

@flow(name="calculate_position_size", retries=1)
def calculate_position_size(symbol: str, signal_strength: float):
    """Position sizing based on risk parameters"""
    # Load risk parameters
    # Calculate portfolio value
    # Determine position size
    # Validate against limits

@flow(name="validate_risk_limits", retries=1)
def validate_risk_limits(proposed_trade: Dict):
    """Risk limit validation before trade execution"""
    # Check position limits
    # Validate exposure limits
    # Check drawdown limits
    # Approve or reject trade

@flow(name="monitor_portfolio_risk", schedule=CronSchedule(cron="*/15 * * * *"))
def monitor_portfolio_risk():
    """Continuous portfolio risk monitoring"""
    # Calculate current exposure
    # Check risk metrics
    # Generate alerts if needed
    # Update risk dashboard

@flow(name="generate_risk_alerts", retries=2)
def generate_risk_alerts(risk_event: Dict):
    """Risk alert generation and notification"""
    # Format risk message
    # Send email alerts
    # Update risk dashboard
    # Log risk events

@flow(name="calculate_performance", retries=1, timeout_seconds=600)
def calculate_performance(strategy_id: str, period: str):
    """Performance metrics calculation"""
    # Fetch trade data
    # Calculate returns
    # Compute Sharpe ratio
    # Calculate drawdown
    # Store performance metrics

@flow(name="generate_reports", schedule=CronSchedule(cron="0 18 * * *"))
def generate_reports():
    """Daily performance report generation"""
    # Calculate daily metrics
    # Generate performance charts
    # Create summary report
    # Send email reports

@flow(name="run_backtest", timeout_seconds=3600)
def run_backtest(strategy_config: Dict, start_date: datetime, end_date: datetime):
    """Comprehensive strategy backtesting"""
    # Load historical data
    # Run strategy simulation
    # Calculate performance metrics
    # Generate detailed report
    # Store backtest results

@flow(name="analyze_trades", retries=1)
def analyze_trades(strategy_id: str):
    """Trade pattern analysis"""
    # Fetch trade history
    # Analyze win/loss patterns
    # Calculate trade statistics
    # Identify improvement areas

@flow(name="send_trade_alerts", retries=3)
def send_trade_alerts(trade_data: Dict):
    """Trade execution notifications"""
    # Format trade message
    # Send email notification
    # Update dashboard
    # Log notification

@flow(name="monitor_system_health", schedule=CronSchedule(cron="*/5 * * * *"))
def monitor_system_health():
    """System health monitoring and alerting"""
    # Check service status
    # Monitor database connections
    # Check API availability
    # Send alerts if issues found

@flow(name="aggregate_logs", schedule=CronSchedule(cron="0 1 * * *"))
def aggregate_logs():
    """Daily log aggregation and analysis"""
    # Collect logs from all services
    # Analyze error patterns
    # Generate log summary
    # Clean up old logs

@flow(name="send_daily_summary", schedule=CronSchedule(cron="0 19 * * *"))
def send_daily_summary():
    """Daily trading summary email"""
    # Calculate daily P&L
    # Summarize trades
    # Generate performance metrics
    # Send summary email

@flow(name="trading_pipeline")
def trading_pipeline():
    """Complete trading pipeline execution"""
    # Step 1: Fetch market data
    market_data = fetch_market_data(symbols=["AAPL", "GOOGL"])

    # Step 2: Run strategies
    signals = run_strategy("momentum_strategy", ["AAPL", "GOOGL"])

    # Step 3: Validate risk
    approved_trades = validate_risk_limits(signals)

    # Step 4: Execute trades
    if approved_trades:
        execute_trades(approved_trades)

    # Step 5: Update positions
    update_positions()

    # Step 6: Send notifications
    send_trade_alerts(approved_trades)

@flow(name="parallel_analytics")
def parallel_analytics():
    """Run analytics flows in parallel"""
    with Flow("parallel_analytics") as flow:
        # Run these flows in parallel
        performance_task = calculate_performance.submit("strategy_1", "1M")
        backtest_task = run_backtest.submit(strategy_config, start_date, end_date)
        analysis_task = analyze_trades.submit("strategy_1")

        # Wait for all to complete
        results = [performance_task, backtest_task, analysis_task]
        return results

@flow(name="conditional_trading")
def conditional_trading(market_condition: str):
    """Conditional trading based on market conditions"""
    if market_condition == "volatile":
        # Run conservative strategy
        run_strategy("conservative_strategy")
    elif market_condition == "stable":
        # Run aggressive strategy
        run_strategy("aggressive_strategy")
    else:
        # Run default strategy
        run_strategy("default_strategy")

┌─────────────────────────────────────────────────────────┐
│                Prefect Server                          │
│                (Port 4200)                             │
├─────────────────────────────────────────────────────────┤
│  API Server    │  UI Server    │  Database             │
│  ├── REST API  │  ├── Web UI   │  ├── Flow Runs        │
│  ├── GraphQL   │  ├── Dashboard│  ├── Task Runs        │
│  └── WebSocket │  └── Explorer │  ├── Deployments      │
│                                     ├── Work Pools      │
│                                     └── Blocks          │
└─────────────────────────────────────────────────────────┘

# Start Prefect server locally
prefect server start --host 0.0.0.0 --port 4200

# Configure database
prefect config set PREFECT_API_DATABASE_CONNECTION_URL="postgresql+asyncpg://postgres:password@localhost:5432/prefect"

# Initialize database
prefect database upgrade

# prefect.yaml
prefect:
  api:
    database:
      connection_url: "postgresql+asyncpg://postgres:password@localhost:5432/prefect"
      echo: false
      pool_size: 10
      max_overflow: 20
      pool_pre_ping: true
      pool_recycle: 3600

  server:
    host: "0.0.0.0"
    port: 4200
    ui:
      enabled: true
      host: "0.0.0.0"
      port: 4200

  work_pools:
    - name: "trading-pool"
      type: "process"
      base_job_template:
        job_configuration:
          command: "python -m src.services.{service}.main"
          env:
            PREFECT_API_URL: "http://localhost:4200/api"
            POSTGRES_HOST: "localhost"
            REDIS_HOST: "localhost"

# Prefect Configuration
PREFECT_API_URL=http://localhost:4200/api
PREFECT_API_DATABASE_CONNECTION_URL=postgresql+asyncpg://postgres:password@localhost:5432/prefect
PREFECT_LOGGING_LEVEL=INFO
PREFECT_LOGGING_TO_API_ENABLED=true
PREFECT_SERVER_API_HOST=0.0.0.0
PREFECT_UI_URL=http://localhost:4200

# src/shared/prefect/work_pools.py
from prefect import get_client
from prefect.client.schemas.actions import WorkPoolCreate

async def setup_work_pools():
    """Setup work pools for trading system services"""
    async with get_client() as client:
        # Data Ingestion Pool
        await client.create_work_pool(
            WorkPoolCreate(
                name="data-ingestion-pool",
                type="process",
                base_job_template={
                    "job_configuration": {
                        "command": "python -m src.services.data_ingestion.main",
                        "env": {
                            "PREFECT_API_URL": "http://localhost:4200/api",
                            "SERVICE_NAME": "data_ingestion"
                        }
                    }
                }
            )
        )

        # Strategy Engine Pool
        await client.create_work_pool(
            WorkPoolCreate(
                name="strategy-engine-pool",
                type="process",
                base_job_template={
                    "job_configuration": {
                        "command": "python -m src.services.strategy_engine.main",
                        "env": {
                            "PREFECT_API_URL": "http://localhost:4200/api",
                            "SERVICE_NAME": "strategy_engine"
                        }
                    }
                }
            )
        )

        # Execution Pool
        await client.create_work_pool(
            WorkPoolCreate(
                name="execution-pool",
                type="process",
                base_job_template={
                    "job_configuration": {
                        "command": "python -m src.services.execution.main",
                        "env": {
                            "PREFECT_API_URL": "http://localhost:4200/api",
                            "SERVICE_NAME": "execution"
                        }
                    }
                }
            )
        )

# src/shared/prefect/deployments.py
from prefect import flow
from prefect.deployments import Deployment
from prefect.server.schemas.schedules import CronSchedule

def deploy_trading_flows():
    """Deploy all trading system flows"""

    # Data Ingestion Flows
    fetch_market_data_deployment = Deployment.build_from_flow(
        flow=fetch_market_data,
        name="fetch-market-data",
        schedule=CronSchedule(cron="0 * * * *"),  # Every hour
        work_pool_name="data-ingestion-pool",
        tags=["data-ingestion", "scheduled"]
    )

    # Strategy Engine Flows (Prefect 3.x from_source pattern)
    pairs_deployment = await intraday_pairs_flow.from_source(
        source=str(project_root),
        entrypoint="src/shared/prefect/flows/strategy_engine/pairs_flow.py:intraday_pairs_flow",
    )
    await pairs_deployment.deploy(
        name="Intraday Pairs Trading",
        work_pool_name=PrefectConfig.get_work_pool_name(),
        cron="0 14-21 * * 1-5",  # hourly 9 AM–5 PM ET Mon–Fri
        tags=["strategy-engine", "pairs-trading", "scheduled"],
    )

    # Execution Flows
    execute_trades_deployment = Deployment.build_from_flow(
        flow=execute_trades,
        name="execute-trades",
        work_pool_name="execution-pool",
        tags=["execution", "triggered"]
    )

    # Deploy all flows
    fetch_market_data_deployment.apply()
    run_strategy_deployment.apply()
    execute_trades_deployment.apply()

# src/shared/prefect/monitoring.py
from prefect import get_client
from prefect.events import emit_event

class PrefectMonitoring:
    def __init__(self):
        self.client = get_client()

    async def monitor_flow_runs(self):
        """Monitor flow run status and performance"""
        async with self.client:
            # Get recent flow runs
            flow_runs = await self.client.read_flow_runs(
                limit=100,
                sort="START_TIME_DESC"
            )

            # Check for failed runs
            failed_runs = [run for run in flow_runs if run.state_type == "FAILED"]

            if failed_runs:
                await self.handle_failed_runs(failed_runs)

    async def handle_failed_runs(self, failed_runs):
        """Handle failed flow runs"""
        for run in failed_runs:
            # Emit failure event
            await emit_event(
                event="flow-run-failed",
                resource={"prefect.resource.id": f"prefect.flow-run.{run.id}"},
                payload={
                    "flow_name": run.name,
                    "error_message": run.state.message,
                    "start_time": run.start_time,
                    "end_time": run.end_time
                }
            )

-- Create Prefect database user
CREATE USER prefect_user WITH PASSWORD 'prefect_password';

-- Grant permissions
GRANT CONNECT ON DATABASE prefect TO prefect_user;
GRANT USAGE ON SCHEMA public TO prefect_user;
GRANT CREATE ON SCHEMA public TO prefect_user;

-- Grant table permissions
GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public TO prefect_user;
GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public TO prefect_user;

# src/shared/prefect/security.py
from prefect.server.api.server import create_app
from prefect.server.database import get_database_engine
from prefect.settings import PREFECT_API_KEY

def configure_prefect_security():
    """Configure Prefect security settings"""

    # API Key authentication
    if PREFECT_API_KEY:
        app = create_app()
        app.add_middleware(
            APIKeyMiddleware,
            api_key=PREFECT_API_KEY
        )

    # CORS configuration
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["http://localhost:3000", "http://localhost:8000"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

#!/bin/bash
# start_prefect.sh

echo "Starting Prefect Server..."

# Check if PostgreSQL is running
if ! pg_isready -h localhost -p 5432; then
    echo "PostgreSQL is not running. Please start PostgreSQL first."
    exit 1
fi

# Check if Prefect database exists
if ! psql -h localhost -U postgres -lqt | cut -d \| -f 1 | grep -qw prefect; then
    echo "Creating Prefect database..."
    createdb -h localhost -U postgres prefect
fi

# Configure Prefect
echo "Configuring Prefect..."
prefect config set PREFECT_API_DATABASE_CONNECTION_URL="postgresql+asyncpg://postgres:password@localhost:5432/prefect"
prefect config set PREFECT_API_URL="http://localhost:4200/api"

# Initialize database
echo "Initializing Prefect database..."
prefect database upgrade

# Start Prefect server
echo "Starting Prefect server..."
prefect server start --host 0.0.0.0 --port 4200 --log-level INFO

#!/bin/bash
# start_prefect_worker.sh

echo "Starting Prefect Worker..."

# Check if Prefect server is running
if ! curl -s http://localhost:4200/api/health > /dev/null; then
    echo "Prefect server is not running. Please start Prefect server first."
    exit 1
fi

# Start worker
echo "Starting Prefect worker..."
prefect worker start --pool trading-pool --limit 10

# Check database connection
psql -h localhost -U postgres -d prefect -c "SELECT 1;"

# Check Prefect database tables
psql -h localhost -U postgres -d prefect -c "\dt"

# Check flow deployments
prefect deployment ls

# Check work pools
prefect work-pool ls

# Check workers
prefect worker ls

# Check flow runs
prefect flow-run ls

# Check specific flow run
prefect flow-run inspect <flow-run-id>

# Check flow run logs
prefect flow-run logs <flow-run-id>

-- Create indexes for Prefect tables
CREATE INDEX idx_flow_runs_state_type ON flow_run(state_type);
CREATE INDEX idx_flow_runs_start_time ON flow_run(start_time);
CREATE INDEX idx_task_runs_state_type ON task_run(state_type);
CREATE INDEX idx_task_runs_start_time ON task_run(start_time);

# src/shared/prefect/connection_pool.py
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

def create_prefect_engine():
    """Create optimized Prefect database engine"""
    return create_engine(
        "postgresql+asyncpg://postgres:password@localhost:5432/prefect",
        poolclass=QueuePool,
        pool_size=10,
        max_overflow=20,
        pool_pre_ping=True,
        pool_recycle=3600,
        echo=False
    )

#!/bin/bash
# backup_prefect.sh

echo "Backing up Prefect database..."

# Create backup
pg_dump -h localhost -U postgres -d prefect > prefect_backup_$(date +%Y%m%d_%H%M%S).sql

echo "Prefect database backup completed."

#!/bin/bash
# backup_prefect_config.sh

echo "Backing up Prefect configuration..."

# Export flows
prefect flow ls --format json > flows_backup_$(date +%Y%m%d_%H%M%S).json

# Export deployments
prefect deployment ls --format json > deployments_backup_$(date +%Y%m%d_%H%M%S).json

echo "Prefect configuration backup completed."

src/
├── shared/
│   └── prefect/
│       ├── __init__.py
│       ├── config.py              # Prefect configuration
│       ├── flows/
│       │   ├── __init__.py
│       │   ├── data_ingestion/
│       │   │   ├── __init__.py
│       │   │   ├── polygon_flows.py      # Polygon.io data flows
│       │   │   ├── yahoo_flows.py        # Yahoo Finance flows
│       │   │   └── validation_flows.py   # Data quality checks
│       │   ├── analytics/
│       │   │   ├── __init__.py
│       │   │   └── indicator_flows.py    # Technical indicators
│       │   └── maintenance/
│       │       ├── __init__.py
│       │       └── cleanup_flows.py      # Data archival, cleanup
│       ├── deployments/
│       │   ├── __init__.py
│       │   └── deployments.py            # Deployment definitions
│       ├── tasks/
│       │   ├── __init__.py
│       │   ├── data_ingestion.py         # Reusable data ingestion tasks
│       │   └── validation.py             # Data validation tasks
│       └── utils/
│           ├── __init__.py
│           └── helpers.py                # Prefect utilities
└── ...

scripts/
├── prefect/
│   ├── start_prefect_server.py          # Server startup script
│   ├── start_prefect_worker.py          # Worker startup script
│   └── deploy_flows.py                  # Deployment script
└── ...

# Prefect Configuration
PREFECT_API_URL=http://localhost:4200/api
PREFECT_API_DATABASE_CONNECTION_URL=postgresql+asyncpg://postgres:password@localhost:5432/prefect
PREFECT_LOGGING_LEVEL=INFO
PREFECT_LOGGING_TO_API_ENABLED=true
PREFECT_SERVER_API_HOST=0.0.0.0
PREFECT_UI_URL=http://localhost:4200

# Prefect Work Pools
PREFECT_WORK_POOL_DATA_INGESTION=default-agent-pool
PREFECT_WORK_POOL_ANALYTICS=default-agent-pool
PREFECT_WORK_POOL_MAINTENANCE=default-agent-pool

# Prefect Flow Configuration
PREFECT_FLOW_RETRY_ATTEMPTS=3
PREFECT_FLOW_RETRY_DELAY_SECONDS=60
PREFECT_FLOW_TIMEOUT_SECONDS=3600

# Prefect Configuration
prefect_api_url: str = Field(
    default="http://localhost:4200/api",
    alias="PREFECT_API_URL"
)
prefect_db_url: str = Field(
    default="postgresql+asyncpg://postgres:password@localhost:5432/prefect",
    alias="PREFECT_API_DATABASE_CONNECTION_URL"
)
prefect_logging_level: str = Field(
    default="INFO",
    alias="PREFECT_LOGGING_LEVEL"
)

from prefect import flow, task
from prefect.tasks import task_input_hash
from datetime import timedelta

@task(
    retries=3,
    retry_delay_seconds=60,
    cache_key_fn=task_input_hash,
    cache_expiration=timedelta(hours=1)
)
async def load_symbol_data_task(symbol: str, days_back: int):
    """Load data for a single symbol"""
    # Implementation using existing HistoricalDataLoader
    pass

@flow(
    name="polygon-daily-ingestion",
    retries=2,
    timeout_seconds=3600
)
async def polygon_daily_ingestion():
    """Daily end-of-day data ingestion from Polygon.io"""
    # Get active symbols
    # Load data for each symbol
    # Update load_runs tracking
    # Validate data quality
    pass

@flow(name="data-ingestion-pipeline")
async def data_ingestion_pipeline():
    """Complete data ingestion pipeline"""
    # Step 1: Load Polygon data
    polygon_result = await polygon_daily_ingestion()

    # Step 2: Load Yahoo data (after Polygon completes)
    yahoo_result = await yahoo_daily_ingestion()

    # Step 3: Calculate indicators (after data loads complete)
    indicator_result = await calculate_daily_indicators()

    # Step 4: Validate data quality
    validation_result = await validate_data_quality()

    return {
        "polygon": polygon_result,
        "yahoo": yahoo_result,
        "indicators": indicator_result,
        "validation": validation_result
    }

Data Ingestion → Strategy Engine → Risk Management → Execution
     ↓                ↓                ↓              ↓
Analytics Service ← Notification Service ← Redis ← PostgreSQL

Market Data Flow → Strategy Flow → Risk Flow → Execution Flow
       ↓               ↓            ↓           ↓
   Analytics Flow ← Notification Flow ← Monitoring Flow

# Create conda environment
conda create -n trading-system python=3.11
conda activate trading-system

# Install dependencies
conda install -c conda-forge postgresql redis
pip install -r requirements.txt

# Setup databases
createdb trading_system
redis-server

# Pre-commit hooks
pre-commit install

# Code formatting
black .
isort .

# Linting
flake8 .

# Type checking
mypy .