TradAI Final Architecture - Services¶

# services/api_gateway/api/routes/backtest.py
from fastapi import APIRouter, HTTPException, Depends
import uuid

router = APIRouter(prefix="/backtest", tags=["backtest"])

@router.post("", response_model=BacktestResponse)
async def trigger_backtest(
    request: BacktestRequest,
    sqs: SQSClient = Depends(),
    dynamo: DynamoDBClient = Depends()
):
    """
    Trigger a new backtest workflow.
    Returns immediately with run_id for status polling.
    """
    run_id = str(uuid.uuid4())

    # Register in DynamoDB
    await dynamo.create_workflow(
        run_id=run_id,
        workflow_type="backtest",
        status="QUEUED",
        metadata={
            "strategy_name": request.strategy_name,
            "strategy_version": request.strategy_version,
            "experiment_name": request.experiment_name
        }
    )

    # Send to SQS (async processing)
    await sqs.send_message(
        queue_url=settings.BACKTEST_QUEUE_URL,
        message={"run_id": run_id, **request.dict()},
        message_group_id=request.strategy_name,
        deduplication_id=run_id
    )

    return BacktestResponse(
        run_id=run_id,
        status="QUEUED",
        status_url=f"/backtest/{run_id}/status"
    )


@router.get("/{run_id}/status", response_model=BacktestStatus)
async def get_backtest_status(
    run_id: str,
    dynamo: DynamoDBClient = Depends()
):
    """Get backtest status and results."""
    workflow = await dynamo.get_workflow(run_id)

    if not workflow:
        raise HTTPException(status_code=404, detail="Backtest not found")

    response = BacktestStatus(
        run_id=run_id,
        status=workflow["status"],
        created_at=workflow["created_at"],
        updated_at=workflow["updated_at"]
    )

    if workflow["status"] == "COMPLETED":
        response.result = workflow.get("result")
        response.mlflow_url = f"{settings.MLFLOW_URL}/experiments/{run_id}"

    return response

# services/api_gateway/api/routes/health.py
from fastapi import APIRouter, Response
import httpx

router = APIRouter(tags=["health"])

@router.get("/health")
async def health_check():
    """Basic liveness check."""
    return {"status": "healthy", "service": "backend-api"}


@router.get("/ready")
async def readiness_check():
    """Readiness check - are dependencies available?"""
    checks = {"mlflow": False, "data_collection": False, "dynamodb": False}

    # Check MLflow
    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            resp = await client.get(f"{settings.MLFLOW_URL}/health")
            checks["mlflow"] = resp.status_code == 200
    except:
        pass

    # Check Data Collection
    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            resp = await client.get(f"{settings.DATA_COLLECTION_URL}/health")
            checks["data_collection"] = resp.status_code == 200
    except:
        pass

    # Check DynamoDB
    try:
        dynamo = DynamoDBClient()
        await dynamo.health_check()
        checks["dynamodb"] = True
    except:
        pass

    all_healthy = all(checks.values())
    return Response(
        content=json.dumps({"status": "ready" if all_healthy else "not_ready", "checks": checks}),
        status_code=200 if all_healthy else 503,
        media_type="application/json"
    )

# services/data_collection/api/routes/data.py
from fastapi import APIRouter
from pydantic import BaseModel
from datetime import datetime
import arcticdb as adb

router = APIRouter(prefix="/data", tags=["data"])

class FreshnessRequest(BaseModel):
    symbols: list[str]
    timeframe: str

class FreshnessResponse(BaseModel):
    results: dict[str, dict]
    all_fresh: bool

@router.post("/freshness", response_model=FreshnessResponse)
async def check_data_freshness(request: FreshnessRequest):
    """
    Check if data is fresh enough for backtesting.
    Fresh = last update within 24 hours.
    """
    arctic = adb.Arctic(f"s3://{settings.ARCTIC_S3_BUCKET}")
    library = arctic.get_library(settings.ARCTIC_LIBRARY)

    results = {}
    all_fresh = True

    for symbol in request.symbols:
        symbol_key = f"{symbol}_{request.timeframe}"

        try:
            info = library.get_info(symbol_key)
            last_update = info.get("last_update", datetime.min)
            age_hours = (datetime.utcnow() - last_update).total_seconds() / 3600

            is_fresh = age_hours < 24
            results[symbol] = {
                "last_update": last_update.isoformat(),
                "age_hours": round(age_hours, 2),
                "is_fresh": is_fresh,
                "row_count": info.get("row_count", 0)
            }

            if not is_fresh:
                all_fresh = False

        except Exception as e:
            results[symbol] = {"error": str(e), "is_fresh": False}
            all_fresh = False

    return FreshnessResponse(results=results, all_fresh=all_fresh)

# services/strategy_service/commands/prepare_config.py
import boto3
import mlflow
from omegaconf import OmegaConf
import yaml
import os

def prepare_config(input_data: dict) -> dict:
    """
    Prepare configuration for backtest.
    1. Fetch strategy metadata from MLflow
    2. Download base config from S3
    3. Merge with overrides
    4. Upload merged config to S3
    """
    s3 = boto3.client("s3")

    # Get strategy from MLflow
    mlflow.set_tracking_uri(os.environ["MLFLOW_TRACKING_URI"])
    client = mlflow.tracking.MlflowClient()

    model_version = client.get_model_version(
        name=input_data["strategy_name"],
        version=input_data["strategy_version"]
    )

    ecr_url = model_version.tags.get("ecr_url")
    base_config_path = model_version.tags.get("config_path")

    # Download base config
    bucket, key = base_config_path.replace("s3://", "").split("/", 1)
    response = s3.get_object(Bucket=bucket, Key=key)
    base_config = OmegaConf.create(yaml.safe_load(response["Body"]))

    # Merge with overrides
    overrides = OmegaConf.create(input_data.get("config_overrides", {}))
    merged_config = OmegaConf.merge(base_config, overrides)
    merged_config.run_id = input_data["run_id"]
    merged_config.experiment_name = input_data["experiment_name"]

    # Upload merged config
    temp_key = f"temp/{input_data['run_id']}/config.yaml"
    s3.put_object(
        Bucket=os.environ["S3_CONFIG_BUCKET"],
        Key=temp_key,
        Body=OmegaConf.to_yaml(merged_config)
    )

    return {
        "ecr_url": ecr_url,
        "config_path": f"s3://{os.environ['S3_CONFIG_BUCKET']}/{temp_key}",
        "strategy_name": input_data["strategy_name"],
        "strategy_version": input_data["strategy_version"]
    }

# services/backend/src/tradai/backend/core/repositories.py

class BacktestExecutor(Protocol):
    """Protocol for backtest execution backends."""

    def submit(self, job: BacktestJobMessage) -> str:
        """
        Submit a backtest job for execution.

        Args:
            job: Backtest job message with config and metadata

        Returns:
            Execution identifier (task ARN, message ID, container ID, etc.)

        Raises:
            ExternalServiceError: If submission fails
        """
        ...

# services/backend/src/tradai/backend/infrastructure/ecs.py

class ECSBacktestExecutor(LoggerMixin):
    """Launches strategy container as ECS Fargate task."""

    def __init__(
        self,
        cluster: str,
        task_definition_prefix: str,
        subnets: list[str],
        security_groups: list[str],
        dynamodb_table: str,
        mlflow_uri: str,
        results_bucket: str,
        ecs_client: Any = None,
    ) -> None:
        super().__init__()
        self._cluster = cluster
        self._task_prefix = task_definition_prefix
        self._subnets = subnets
        self._security_groups = security_groups
        self._table_name = dynamodb_table
        self._mlflow_uri = mlflow_uri
        self._results_bucket = results_bucket
        self._ecs = ecs_client or boto3.client("ecs")

    def submit(self, job: BacktestJobMessage) -> str:
        """Launch ECS task for backtest execution."""
        task_def = f"{self._task_prefix}{job.config.strategy}"

        response = self._ecs.run_task(
            cluster=self._cluster,
            taskDefinition=task_def,
            launchType="FARGATE",
            networkConfiguration={
                "awsvpcConfiguration": {
                    "subnets": self._subnets,
                    "securityGroups": self._security_groups,
                    "assignPublicIp": "DISABLED",
                }
            },
            overrides={
                "containerOverrides": [{
                    "name": "strategy",
                    "command": self._build_command(job),
                    "environment": [
                        {"name": "RUN_ID", "value": job.job_id},
                        {"name": "EXPERIMENT_NAME", "value": job.experiment_name},
                        {"name": "DYNAMODB_TABLE", "value": self._table_name},
                        {"name": "MLFLOW_TRACKING_URI", "value": self._mlflow_uri},
                        {"name": "S3_RESULTS_BUCKET", "value": self._results_bucket},
                    ],
                }],
            },
            capacityProviderStrategy=[
                {"capacityProvider": "FARGATE_SPOT", "weight": 1},
                {"capacityProvider": "FARGATE", "weight": 0, "base": 1},
            ],
        )

        task_arn = response["tasks"][0]["taskArn"]
        self.logger.info(f"Launched ECS task: {task_arn} for job {job.job_id}")
        return task_arn

    def _build_command(self, job: BacktestJobMessage) -> list[str]:
        """Build container entrypoint command."""
        return [
            "python", "/app/entrypoint.py",
            "--strategy", job.config.strategy,
            "--timeframe", job.config.timeframe,
            "--timerange", f"{job.config.start_date}-{job.config.end_date}",
            "--pairs", ",".join(job.config.symbols),
        ]

# services/backend/src/tradai/backend/infrastructure/sqs.py (existing)

class SQSBacktestExecutor(LoggerMixin):
    """Submits backtest jobs to SQS queue for async processing."""

    def __init__(self, sqs_adapter: SQSAdapter) -> None:
        super().__init__()
        self._adapter = sqs_adapter

    def submit(self, job: BacktestJobMessage) -> str:
        """Submit job to SQS queue."""
        self.logger.info(f"Submitting job {job.job_id} to SQS queue")
        message_id = self._adapter.send_backtest_job(job)
        return message_id

# services/backend/src/tradai/backend/infrastructure/local.py

class LocalBacktestExecutor(LoggerMixin):
    """Runs strategy container locally via Docker."""

    def __init__(
        self,
        docker_client: Any = None,
        data_dir: Path | None = None,
        results_dir: Path | None = None,
        mlflow_uri: str = "http://localhost:5000",
    ) -> None:
        super().__init__()
        self._docker = docker_client or docker.from_env()
        self._data_dir = data_dir or Path.home() / ".tradai" / "data"
        self._results_dir = results_dir or Path.home() / ".tradai" / "results"
        self._mlflow_uri = mlflow_uri

    def submit(self, job: BacktestJobMessage) -> str:
        """Run strategy container locally."""
        self._results_dir.mkdir(parents=True, exist_ok=True)

        container = self._docker.containers.run(
            image=f"strategy-{job.config.strategy}:latest",
            command=self._build_command(job),
            environment={
                "RUN_ID": job.job_id,
                "MLFLOW_TRACKING_URI": self._mlflow_uri,
                "EXPERIMENT_NAME": job.experiment_name,
            },
            volumes={
                str(self._data_dir): {"bind": "/data", "mode": "ro"},
                str(self._results_dir / job.job_id): {"bind": "/results", "mode": "rw"},
            },
            detach=True,
            remove=True,
        )

        self.logger.info(f"Started local container: {container.id[:12]} for job {job.job_id}")
        return container.id

# services/backend/src/tradai/backend/infrastructure/stepfunctions.py

class StepFunctionsExecutor(LoggerMixin):
    """Executes backtest via Step Functions state machine."""

    def __init__(
        self,
        state_machine_arn: str,
        sfn_client: Any = None,
    ) -> None:
        super().__init__()
        self._state_machine_arn = state_machine_arn
        self._sfn = sfn_client or boto3.client("stepfunctions")

    def submit(self, job: BacktestJobMessage) -> str:
        """Start Step Functions execution."""
        response = self._sfn.start_execution(
            stateMachineArn=self._state_machine_arn,
            name=f"backtest-{job.job_id}",
            input=json.dumps({
                "run_id": job.job_id,
                "config": job.config.model_dump(),
                "experiment_name": job.experiment_name,
            }),
        )

        execution_arn = response["executionArn"]
        self.logger.info(f"Started Step Functions execution: {execution_arn}")
        return execution_arn

# services/backend/src/tradai/backend/api/dependencies.py

def get_backtest_executor(
    settings: BackendSettings = Depends(get_settings),
) -> BacktestExecutor | None:
    """Get executor based on EXECUTOR_MODE environment variable."""
    mode = settings.executor_mode

    if mode == "local":
        import docker
        return LocalBacktestExecutor(
            docker_client=docker.from_env(),
            mlflow_uri=settings.mlflow_tracking_uri,
        )

    elif mode == "ecs":
        return ECSBacktestExecutor(
            cluster=settings.ecs_cluster,
            task_definition_prefix=settings.ecs_task_prefix,
            subnets=settings.ecs_subnets,
            security_groups=settings.ecs_security_groups,
            dynamodb_table=settings.workflow_state_table,
            mlflow_uri=settings.mlflow_tracking_uri,
            results_bucket=settings.s3_results_bucket,
        )

    elif mode == "sqs":
        sqs_adapter = get_sqs_adapter(settings)
        return SQSBacktestExecutor(sqs_adapter) if sqs_adapter else None

    elif mode == "stepfunctions":
        return StepFunctionsExecutor(
            state_machine_arn=settings.step_functions_arn,
        )

    else:
        # Dev mode - no executor (jobs stay PENDING)
        return None

#!/usr/bin/env python3
"""Strategy container entrypoint - handles backtest execution and status updates."""
# strategies/base/entrypoint.py

import os
import sys
import subprocess
from pathlib import Path

import boto3
from freqtrade.data.btanalysis import load_backtest_stats


def wait_for_mlflow(uri: str, max_retries: int = 10, backoff_base: float = 2.0) -> bool:
    """Wait for MLflow to be healthy with exponential backoff.

    CRITICAL: Prevents container failures when MLflow is not yet ready.

    Args:
        uri: MLflow tracking URI
        max_retries: Maximum number of retry attempts (default: 10 = ~17 min total)
        backoff_base: Base for exponential backoff in seconds

    Returns:
        True if MLflow is healthy, False if all retries exhausted
    """
    import time
    import requests

    health_url = f"{uri.rstrip('/')}/health"

    for attempt in range(max_retries):
        try:
            response = requests.get(health_url, timeout=5)
            if response.status_code == 200:
                print(f"MLflow healthy after {attempt + 1} attempt(s)")
                return True
        except requests.RequestException as e:
            wait_time = backoff_base ** attempt
            print(f"MLflow not ready (attempt {attempt + 1}/{max_retries}): {e}")
            print(f"Retrying in {wait_time:.1f}s...")
            time.sleep(wait_time)

    print(f"MLflow not healthy after {max_retries} attempts")
    return False


def main() -> int:
    """Main entrypoint for strategy container."""
    run_id = os.environ["RUN_ID"]
    strategy = os.environ.get("STRATEGY", os.environ.get("FREQTRADE_STRATEGY"))
    dynamodb_table = os.environ.get("DYNAMODB_TABLE")
    mlflow_uri = os.environ.get("MLFLOW_TRACKING_URI")
    experiment_name = os.environ.get("EXPERIMENT_NAME", f"backtests/{strategy}")

    # Initialize DynamoDB (optional - graceful degradation for local mode)
    dynamodb = None
    if dynamodb_table:
        dynamodb = boto3.resource("dynamodb").Table(dynamodb_table)

    try:
        # 0. CRITICAL: Wait for MLflow to be healthy (prevents race condition)
        if mlflow_uri:
            if not wait_for_mlflow(mlflow_uri):
                raise RuntimeError(f"MLflow at {mlflow_uri} not available after retries")

        # 1. Update status to RUNNING
        if dynamodb:
            dynamodb.update_item(
                Key={"run_id": run_id},
                UpdateExpression="SET #s = :s, updated_at = :t",
                ExpressionAttributeNames={"#s": "status"},
                ExpressionAttributeValues={":s": "RUNNING", ":t": datetime.utcnow().isoformat()},
            )

        # 2. Build and execute Freqtrade command
        command = build_freqtrade_command()
        print(f"Executing: {' '.join(command)}")
        result = subprocess.run(command, check=True, capture_output=True, text=True)
        print(result.stdout)

        # 3. Parse results
        results_dir = Path(os.environ.get("RESULTS_DIR", "/freqtrade/user_data/backtest_results"))
        stats = load_backtest_stats(results_dir)

        # 4. Log to MLflow
        if mlflow_uri:
            import mlflow
            mlflow.set_tracking_uri(mlflow_uri)
            mlflow.set_experiment(experiment_name)
            with mlflow.start_run(run_name=run_id):
                mlflow.log_metrics(extract_metrics(stats))
                mlflow.log_params({"strategy": strategy, "run_id": run_id})

        # 5. Upload results to S3
        upload_results_to_s3(run_id, stats)

        # 6. Update status to COMPLETED
        if dynamodb:
            dynamodb.update_item(
                Key={"run_id": run_id},
                UpdateExpression="SET #s = :s, #r = :r, updated_at = :t",
                ExpressionAttributeNames={"#s": "status", "#r": "result"},
                ExpressionAttributeValues={
                    ":s": "COMPLETED",
                    ":r": json.dumps(summarize_results(stats)),
                    ":t": datetime.utcnow().isoformat(),
                },
            )

        print(f"Backtest completed successfully: {run_id}")
        return 0

    except subprocess.CalledProcessError as e:
        error_msg = f"Freqtrade execution failed: {e.stderr}"
        handle_failure(dynamodb, run_id, error_msg)
        return 1

    except Exception as e:
        error_msg = f"Unexpected error: {str(e)}"
        handle_failure(dynamodb, run_id, error_msg)
        return 1


def build_freqtrade_command() -> list[str]:
    """Build Freqtrade CLI command from environment variables."""
    cmd = ["freqtrade", "backtesting"]

    if strategy := os.environ.get("STRATEGY"):
        cmd.extend(["--strategy", strategy])
    if timeframe := os.environ.get("TIMEFRAME"):
        cmd.extend(["--timeframe", timeframe])
    if timerange := os.environ.get("TIMERANGE"):
        cmd.extend(["--timerange", timerange])
    if pairs := os.environ.get("PAIRS"):
        cmd.extend(["--pairs", pairs])
    if config := os.environ.get("CONFIG_PATH"):
        cmd.extend(["--config", config])

    cmd.extend(["--export", "trades"])
    return cmd


def handle_failure(dynamodb, run_id: str, error_msg: str) -> None:
    """Update status to FAILED."""
    print(error_msg, file=sys.stderr)
    if dynamodb:
        dynamodb.update_item(
            Key={"run_id": run_id},
            UpdateExpression="SET #s = :s, #e = :e, updated_at = :t",
            ExpressionAttributeNames={"#s": "status", "#e": "error"},
            ExpressionAttributeValues={
                ":s": "FAILED",
                ":e": error_msg[:1000],
                ":t": datetime.utcnow().isoformat(),
            },
        )


if __name__ == "__main__":
    sys.exit(main())

# lambda/sqs_consumer/index.py
import json
import os
import boto3
from datetime import datetime

ecs_client = boto3.client('ecs')
sf_client = boto3.client('stepfunctions')
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ['WORKFLOW_STATE_TABLE'])

# Configuration
LAUNCH_MODE = os.environ.get('LAUNCH_MODE', 'ecs')  # 'ecs' or 'stepfunctions'
ECS_CLUSTER = os.environ.get('ECS_CLUSTER')
ECS_TASK_PREFIX = os.environ.get('ECS_TASK_PREFIX', 'strategy-')
ECS_SUBNETS = os.environ.get('ECS_SUBNETS', '').split(',')
ECS_SECURITY_GROUPS = os.environ.get('ECS_SECURITY_GROUPS', '').split(',')
STEP_FUNCTIONS_ARN = os.environ.get('BACKTEST_WORKFLOW_ARN')


def handler(event, context):
    """Process SQS messages and launch backtests."""
    batch_item_failures = []

    for record in event['Records']:
        message_id = record['messageId']

        try:
            message = json.loads(record['body'])
            run_id = message['run_id']

            # Idempotency check
            if is_already_running(run_id):
                print(f"Backtest {run_id} already running, skipping")
                continue

            # Launch based on mode
            if LAUNCH_MODE == 'stepfunctions':
                execution_arn = launch_step_functions(message, run_id)
            else:  # Default: ECS direct
                execution_arn = launch_ecs_task(message, run_id)

            # Store execution ARN
            store_execution_arn(run_id, execution_arn)
            print(f"Launched backtest {run_id}: {execution_arn}")

        except Exception as e:
            print(f"Error processing {message_id}: {str(e)}")
            batch_item_failures.append({"itemIdentifier": message_id})

    return {"batchItemFailures": batch_item_failures}


def launch_ecs_task(message: dict, run_id: str) -> str:
    """Launch ECS task directly (simpler, lower latency)."""
    strategy = message['config']['strategy']
    task_def = f"{ECS_TASK_PREFIX}{strategy}"

    response = ecs_client.run_task(
        cluster=ECS_CLUSTER,
        taskDefinition=task_def,
        launchType='FARGATE',
        networkConfiguration={
            'awsvpcConfiguration': {
                'subnets': ECS_SUBNETS,
                'securityGroups': ECS_SECURITY_GROUPS,
                'assignPublicIp': 'DISABLED',
            }
        },
        overrides={
            'containerOverrides': [{
                'name': 'strategy',
                'environment': [
                    {'name': 'RUN_ID', 'value': run_id},
                    {'name': 'STRATEGY', 'value': strategy},
                    {'name': 'TIMEFRAME', 'value': message['config']['timeframe']},
                    {'name': 'TIMERANGE', 'value': f"{message['config']['start_date']}-{message['config']['end_date']}"},
                    {'name': 'PAIRS', 'value': ','.join(message['config']['symbols'])},
                    {'name': 'EXPERIMENT_NAME', 'value': message.get('experiment_name', f'backtests/{strategy}')},
                    {'name': 'DYNAMODB_TABLE', 'value': os.environ['WORKFLOW_STATE_TABLE']},
                    {'name': 'MLFLOW_TRACKING_URI', 'value': os.environ.get('MLFLOW_TRACKING_URI', '')},
                ],
            }],
        },
        capacityProviderStrategy=[
            {'capacityProvider': 'FARGATE_SPOT', 'weight': 1},
            {'capacityProvider': 'FARGATE', 'weight': 0, 'base': 1},
        ],
    )

    return response['tasks'][0]['taskArn']


def launch_step_functions(message: dict, run_id: str) -> str:
    """Launch Step Functions execution (for complex workflows)."""
    response = sf_client.start_execution(
        stateMachineArn=STEP_FUNCTIONS_ARN,
        name=f"backtest-{run_id}",
        input=json.dumps(message)
    )
    return response['executionArn']


def is_already_running(run_id: str) -> bool:
    response = table.get_item(Key={'run_id': run_id})
    return response.get('Item', {}).get('status') == 'RUNNING'


def store_execution_arn(run_id: str, execution_arn: str) -> None:
    table.update_item(
        Key={'run_id': run_id},
        UpdateExpression='SET execution_arn = :arn, updated_at = :t',
        ExpressionAttributeValues={
            ':arn': execution_arn,
            ':t': datetime.utcnow().isoformat(),
        },
    )

# lambda/data_collection_proxy/index.py
import json
import httpx
import os

DATA_COLLECTION_URL = os.environ.get(
    'DATA_COLLECTION_URL',
    'http://data-collection.tradai.local:8002'
)

def handler(event, context):
    """Proxy requests to Data Collection Service (VPC internal)."""
    operation = event.get('operation')

    with httpx.Client(timeout=30.0) as client:
        if operation == 'check-freshness':
            response = client.post(
                f"{DATA_COLLECTION_URL}/data/freshness",
                json={"symbols": event['symbols'], "timeframe": event['timeframe']}
            )
        elif operation == 'validate':
            response = client.post(
                f"{DATA_COLLECTION_URL}/data/validate",
                json={"symbols": event['symbols']}
            )
        else:
            raise ValueError(f"Unknown operation: {operation}")

        response.raise_for_status()
        return response.json()

# libs/common/src/circuit_breaker.py
from enum import Enum
from datetime import datetime
from typing import Callable
import functools

class CircuitState(Enum):
    CLOSED = "closed"       # Normal
    OPEN = "open"           # Failing
    HALF_OPEN = "half_open" # Testing

class CircuitBreakerOpenError(Exception):
    pass

class CircuitBreaker:
    """
    Circuit breaker for resilient service calls.

    Usage:
        cb = CircuitBreaker(failure_threshold=5, recovery_timeout=30)

        @cb
        async def call_external_service():
            ...
    """
    def __init__(self, failure_threshold=5, recovery_timeout=30):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = CircuitState.CLOSED
        self.failures = 0
        self.last_failure_time = None

    def __call__(self, func: Callable):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            self._check_state()
            try:
                result = await func(*args, **kwargs)
                self._on_success()
                return result
            except Exception as e:
                self._on_failure()
                raise
        return wrapper

    def _check_state(self):
        if self.state == CircuitState.OPEN:
            if self._should_attempt_reset():
                self.state = CircuitState.HALF_OPEN
            else:
                raise CircuitBreakerOpenError("Circuit is open")

    def _should_attempt_reset(self) -> bool:
        if not self.last_failure_time:
            return True
        elapsed = (datetime.utcnow() - self.last_failure_time).total_seconds()
        return elapsed >= self.recovery_timeout

    def _on_success(self):
        self.failures = 0
        self.state = CircuitState.CLOSED

    def _on_failure(self):
        self.failures += 1
        self.last_failure_time = datetime.utcnow()
        if self.failures >= self.failure_threshold:
            self.state = CircuitState.OPEN

# libs/tradai-common/src/tradai/common/

# Base service class
class BaseService(LoggerMixin):
    """Base class with logging and Hydra config support."""
    settings_class: type[Settings]

    @classmethod
    def from_hydra_cfg(cls, cfg: DictConfig, config_name: str) -> Self:
        """Create service from Hydra configuration."""

# Base settings class
class Settings(BaseSettings):
    """Pydantic settings with Hydra integration."""

    @classmethod
    def from_hydra_cfg(cls, cfg: DictConfig, config_name: str) -> Self:
        """Load settings from Hydra config."""

# Logger mixin
class LoggerMixin:
    """Provides structured JSON logging."""

    @property
    def logger(self) -> logging.Logger:
        """Get configured logger for this class."""

# libs/tradai-{service}/src/tradai/{service}/__main__.py
"""Entry point for: python -m tradai.{service}"""

from tradai.{service}.cli import app

if __name__ == "__main__":
    app()

# libs/tradai-{service}/src/tradai/{service}/cli.py
"""CLI commands for {service}."""

import typer
import uvicorn
from typing import Optional

from tradai.{service}.config import Settings
from tradai.{service}.app import create_app

app = typer.Typer(
    name="tradai-{service}",
    help="TradAI {Service} - {description}",
    add_completion=False,
)


@app.command()
def serve(
    host: str = typer.Option("0.0.0.0", "--host", "-h", help="Bind host"),
    port: int = typer.Option(8000, "--port", "-p", help="Bind port"),
    reload: bool = typer.Option(False, "--reload", "-r", help="Enable hot reload"),
    workers: int = typer.Option(1, "--workers", "-w", help="Number of workers"),
    log_level: str = typer.Option("info", "--log-level", "-l", help="Log level"),
):
    """Start the service HTTP server."""
    typer.echo(f"Starting {service} on {host}:{port}")

    uvicorn.run(
        "tradai.{service}.app:create_app",
        factory=True,
        host=host,
        port=port,
        reload=reload,
        workers=workers,
        log_level=log_level,
    )


@app.command()
def health():
    """Check service health (without starting server)."""
    from tradai.{service}.core.services import HealthService

    settings = Settings()
    health_service = HealthService(settings)
    result = health_service.check()

    if result.healthy:
        typer.echo("✓ Service is healthy")
        raise typer.Exit(0)
    else:
        typer.echo(f"✗ Service unhealthy: {result.message}")
        raise typer.Exit(1)


@app.command()
def config():
    """Show current configuration."""
    settings = Settings()
    typer.echo(settings.model_dump_json(indent=2))


@app.command()
def version():
    """Show version information."""
    from tradai.{service} import __version__
    typer.echo(f"tradai-{service} {__version__}")


if __name__ == "__main__":
    app()

# src/tradai_cli/__main__.py
"""Unified TradAI CLI."""

import typer

app = typer.Typer(
    name="tradai",
    help="TradAI Platform CLI",
    add_completion=False,
)

# Import and register sub-apps
from tradai.data.cli import app as data_app
from tradai.backend.cli import app as backend_app

app.add_typer(data_app, name="data", help="Data Collection service")
app.add_typer(backend_app, name="backend", help="Backend API service")


@app.command()
def up(
    services: list[str] = typer.Argument(None, help="Services to start (default: all)"),
    detach: bool = typer.Option(False, "--detach", "-d", help="Run in background"),
):
    """Start services using Docker Compose."""
    import subprocess

    cmd = ["docker", "compose", "up"]
    if detach:
        cmd.append("-d")
    if services:
        cmd.extend(services)

    subprocess.run(cmd)


@app.command()
def down():
    """Stop all services."""
    import subprocess
    subprocess.run(["docker", "compose", "down"])


@app.command()
def logs(
    service: str = typer.Argument(..., help="Service name"),
    follow: bool = typer.Option(False, "--follow", "-f", help="Follow logs"),
):
    """View service logs."""
    import subprocess

    cmd = ["docker", "compose", "logs"]
    if follow:
        cmd.append("-f")
    cmd.append(service)

    subprocess.run(cmd)


if __name__ == "__main__":
    app()