Leaderboard endpoint for PTM

{
  "cost": {
    "input_tokens": 47,
    "estimated_usd": 0.076146,
    "output_tokens": 5067
  },
  "model": "claude-sonnet-4-20250514",
  "status": "success",
  "transport": "anthropic_api",
  "latency_ms": 55240,
  "response_text": "I'll create a comprehensive leaderboard endpoint for PTM (Predictive Trading Models) that returns ranked models with performance metrics and trend indicators.\n\n```python\nfrom flask import Flask, request, jsonify\nfrom flask_sqlalchemy import SQLAlchemy\nfrom sqlalchemy import func, desc, asc, case, and_\nfrom datetime import datetime, timedelta\nfrom typing import Dict, List, Optional, Tuple\nfrom enum import Enum\nimport numpy as np\nfrom dataclasses import dataclass\n\napp = Flask(__name__)\n\nclass PositionType(Enum):\n    LONG = \"long\"\n    SHORT = \"short\"\n    BOTH = \"both\"\n\nclass TrendDirection(Enum):\n    UP = \"up\"\n    DOWN = \"down\"\n    STABLE = \"stable\"\n\n@dataclass\nclass TrendIndicator:\n    direction: TrendDirection\n    magnitude: float  # Percentage change\n    period_days: int\n\n@dataclass\nclass LeaderboardEntry:\n    model_id: str\n    model_name: str\n    rank: int\n    position_type: PositionType\n    trust_score: float\n    sample_size: int\n    win_rate: float\n    avg_return: float\n    sharpe_ratio: float\n    max_drawdown: float\n    total_trades: int\n    trend_indicator: TrendIndicator\n    last_updated: datetime\n\n# Database Models\nclass PTMModel(db.Model):\n    __tablename__ = 'ptm_models'\n    \n    id = db.Column(db.String(36), primary_key=True)\n    name = db.Column(db.String(255), nullable=False)\n    description = db.Column(db.Text)\n    position_type = db.Column(db.Enum(PositionType), nullable=False)\n    created_at = db.Column(db.DateTime, default=datetime.utcnow)\n    is_active = db.Column(db.Boolean, default=True)\n    \n    # Relationships\n    performance_metrics = db.relationship('PTMPerformanceMetric', backref='model', lazy='dynamic')\n    trades = db.relationship('PTMTrade', backref='model', lazy='dynamic')\n\nclass PTMPerformanceMetric(db.Model):\n    __tablename__ = 'ptm_performance_metrics'\n    \n    id = db.Column(db.Integer, primary_key=True)\n    model_id = db.Column(db.String(36), db.ForeignKey('ptm_models.id'), nullable=False)\n    date = db.Column(db.Date, nullable=False)\n    trust_score = db.Column(db.Float, nullable=False)\n    win_rate = db.Column(db.Float, nullable=False)\n    avg_return = db.Column(db.Float, nullable=False)\n    sharpe_ratio = db.Column(db.Float)\n    max_drawdown = db.Column(db.Float)\n    total_trades = db.Column(db.Integer, default=0)\n    sample_size = db.Column(db.Integer, default=0)\n    created_at = db.Column(db.DateTime, default=datetime.utcnow)\n\nclass PTMTrade(db.Model):\n    __tablename__ = 'ptm_trades'\n    \n    id = db.Column(db.Integer, primary_key=True)\n    model_id = db.Column(db.String(36), db.ForeignKey('ptm_models.id'), nullable=False)\n    symbol = db.Column(db.String(20), nullable=False)\n    position_type = db.Column(db.Enum(PositionType), nullable=False)\n    entry_price = db.Column(db.Float, nullable=False)\n    exit_price = db.Column(db.Float)\n    return_pct = db.Column(db.Float)\n    entry_time = db.Column(db.DateTime, nullable=False)\n    exit_time = db.Column(db.DateTime)\n    is_winner = db.Column(db.Boolean)\n    confidence_score = db.Column(db.Float)\n\nclass PTMLeaderboardService:\n    def __init__(self):\n        self.default_lookback_days = 30\n        self.trend_period_days = 7\n        \n    def get_leaderboard(\n        self,\n        position_type: Optional[PositionType] = None,\n        limit: int = 50,\n        min_sample_size: int = 10,\n        lookback_days: int = None\n    ) -> List[LeaderboardEntry]:\n        \"\"\"\n        Generate leaderboard with ranked PTM models\n        \"\"\"\n        if lookback_days is None:\n            lookback_days = self.default_lookback_days\n            \n        cutoff_date = datetime.utcnow() - timedelta(days=lookback_days)\n        \n        # Build base query\n        query = self._build_leaderboard_query(cutoff_date, position_type, min_sample_size)\n        \n        # Execute and process results\n        results = query.limit(limit).all()\n        \n        leaderboard_entries = []\n        for rank, result in enumerate(results, 1):\n            # Get trend indicator\n            trend = self._calculate_trend_indicator(result.model_id, self.trend_period_days)\n            \n            entry = LeaderboardEntry(\n                model_id=result.model_id,\n                model_name=result.model_name,\n                rank=rank,\n                position_type=PositionType(result.position_type),\n                trust_score=round(result.trust_score, 4),\n                sample_size=result.sample_size,\n                win_rate=round(result.win_rate, 4),\n                avg_return=round(result.avg_return, 4),\n                sharpe_ratio=round(result.sharpe_ratio or 0, 4),\n                max_drawdown=round(result.max_drawdown or 0, 4),\n                total_trades=result.total_trades,\n                trend_indicator=trend,\n                last_updated=result.last_updated\n            )\n            \n            leaderboard_entries.append(entry)\n            \n        return leaderboard_entries\n    \n    def _build_leaderboard_query(\n        self,\n        cutoff_date: datetime,\n        position_type: Optional[PositionType],\n        min_sample_size: int\n    ):\n        \"\"\"\n        Build the main leaderboard query with rankings\n        \"\"\"\n        # Subquery for latest metrics per model\n        latest_metrics = db.session.query(\n            PTMPerformanceMetric.model_id,\n            func.max(PTMPerformanceMetric.date).label('latest_date')\n        ).filter(\n            PTMPerformanceMetric.date >= cutoff_date.date()\n        ).group_by(PTMPerformanceMetric.model_id).subquery()\n        \n        # Main query with joins\n        query = db.session.query(\n            PTMModel.id.label('model_id'),\n            PTMModel.name.label('model_name'),\n            PTMModel.position_type,\n            PTMPerformanceMetric.trust_score,\n            PTMPerformanceMetric.win_rate,\n            PTMPerformanceMetric.avg_return,\n            PTMPerformanceMetric.sharpe_ratio,\n            PTMPerformanceMetric.max_drawdown,\n            PTMPerformanceMetric.total_trades,\n            PTMPerformanceMetric.sample_size,\n            PTMPerformanceMetric.created_at.label('last_updated')\n        ).select_from(PTMModel).join(\n            PTMPerformanceMetric,\n            PTMModel.id == PTMPerformanceMetric.model_id\n        ).join(\n            latest_metrics,\n            and_(\n                PTMPerformanceMetric.model_id == latest_metrics.c.model_id,\n                PTMPerformanceMetric.date == latest_metrics.c.latest_date\n            )\n        ).filter(\n            PTMModel.is_active == True,\n            PTMPerformanceMetric.sample_size >= min_sample_size\n        )\n        \n        # Add position type filter if specified\n        if position_type and position_type != PositionType.BOTH:\n            query = query.filter(PTMModel.position_type == position_type)\n        \n        # Order by trust score (primary) and avg return (secondary)\n        query = query.order_by(\n            desc(PTMPerformanceMetric.trust_score),\n            desc(PTMPerformanceMetric.avg_return)\n        )\n        \n        return query\n    \n    def _calculate_trend_indicator(self, model_id: str, period_days: int) -> TrendIndicator:\n        \"\"\"\n        Calculate trend indicator for a model's trust score\n        \"\"\"\n        end_date = datetime.utcnow().date()\n        start_date = end_date - timedelta(days=period_days)\n        \n        # Get trust scores for the period\n        metrics = db.session.query(\n            PTMPerformanceMetric.trust_score,\n            PTMPerformanceMetric.date\n        ).filter(\n            PTMPerformanceMetric.model_id == model_id,\n            PTMPerformanceMetric.date >= start_date,\n            PTMPerformanceMetric.date <= end_date\n        ).order_by(PTMPerformanceMetric.date).all()\n        \n        if len(metrics) < 2:\n            return TrendIndicator(\n                direction=TrendDirection.STABLE,\n                magnitude=0.0,\n                period_days=period_days\n            )\n        \n        # Calculate trend\n        first_score = metrics[0].trust_score\n        last_score = metrics[-1].trust_score\n        \n        if first_score == 0:\n            magnitude = 0.0\n        else:\n            magnitude = ((last_score - first_score) / first_score) * 100\n        \n        # Determine direction\n        if magnitude > 2.0:  # More than 2% improvement\n            direction = TrendDirection.UP\n        elif magnitude < -2.0:  # More than 2% decline\n            direction = TrendDirection.DOWN\n        else:\n            direction = TrendDirection.STABLE\n        \n        return TrendIndicator(\n            direction=direction,\n            magnitude=round(magnitude, 2),\n            period_days=period_days\n        )\n\nclass PTMLeaderboardController:\n    def __init__(self):\n        self.service = PTMLeaderboardService()\n    \n    def get_leaderboard_endpoint(self):\n        \"\"\"\n        GET /api/v1/ptm/leaderboard endpoint handler\n        \"\"\"\n        try:\n            # Parse query parameters\n            position_type_str = request.args.get('position_type', 'both').lower()\n            limit = min(int(request.args.get('limit', 50)), 100)  # Max 100\n            min_sample_size = int(request.args.get('min_sample_size', 10))\n            lookback_days = request.args.get('lookback_days', type=int)\n            include_trends = request.args.get('include_trends', 'true').lower() == 'true'\n            \n            # Validate position type\n            position_type = None\n            if position_type_str in [e.value for e in PositionType]:\n                position_type = PositionType(position_type_str)\n            \n            # Get leaderboard data\n            leaderboard_entries = self.service.get_leaderboard(\n                position_type=position_type,\n                limit=limit,\n                min_sample_size=min_sample_size,\n                lookback_days=lookback_days\n            )\n            \n            # Format response\n            response_data = {\n                'success': True,\n                'data': {\n                    'leaderboard': [\n                        self._format_leaderboard_entry(entry, include_trends)\n                        for entry in leaderboard_entries\n                    ],\n                    'metadata': {\n                        'total_models': len(leaderboard_entries),\n                        'position_type_filter': position_type.value if position_type else 'all',\n                        'min_sample_size': min_sample_size,\n                        'lookback_days': lookback_days or self.service.default_lookback_days,\n                        'last_updated': datetime.utcnow().isoformat()\n                    }\n                }\n            }\n            \n            return jsonify(response_data), 200\n            \n        except ValueError as e:\n            return jsonify({\n                'success': False,\n                'error': {\n                    'code': 'INVALID_PARAMETER',\n                    'message': f'Invalid parameter value: {str(e)}'\n                }\n            }), 400\n            \n        except Exception as e:\n            return jsonify({\n                'success': False,\n                'error': {\n                    'code': 'INTERNAL_ERROR',\n                    'message': 'Failed to retrieve leaderboard data'\n                }\n            }), 500\n    \n    def _format_leaderboard_entry(self, entry: LeaderboardEntry, include_trends: bool) -> Dict:\n        \"\"\"\n        Format leaderboard entry for API response\n        \"\"\"\n        formatted_entry = {\n            'rank': entry.rank,\n            'model': {\n                'id': entry.model_id,\n                'name': entry.model_name,\n                'position_type': entry.position_type.value\n            },\n            'performance': {\n                'trust_score': entry.trust_score,\n                'win_rate': entry.win_rate,\n                'avg_return': entry.avg_return,\n                'sharpe_ratio': entry.sharpe_ratio,\n                'max_drawdown': entry.max_drawdown\n            },\n            'statistics': {\n                'sample_size': entry.sample_size,\n                'total_trades': entry.total_trades\n            },\n            'last_updated': entry.last_updated.isoformat()\n        }\n        \n        # Add trend information if requested\n        if include_trends:\n            formatted_entry['trend'] = {\n                'direction': entry.trend_indicator.direction.value,\n                'magnitude': entry.trend_indicator.magnitude,\n                'period_days': entry.trend_indicator.period_days,\n                'indicator': self._get_trend_emoji(entry.trend_indicator.direction)\n            }\n        \n        return formatted_entry\n    \n    def _get_trend_emoji(self, direction: TrendDirection) -> str:\n        \"\"\"\n        Get emoji indicator for trend direction\n        \"\"\"\n        trend_emojis = {\n            TrendDirection.UP: \"📈\",\n            TrendDirection.DOWN: \"📉\",\n            TrendDirection.STABLE: \"➡️\"\n        }\n        return trend_emojis.get(direction, \"➡️\")\n\n# Initialize controller\nptm_controller = PTMLeaderboardController()\n\n@app.route('/api/v1/ptm/leaderboard', methods=['GET'])\ndef get_ptm_leaderboard():\n    \"\"\"\n    PTM Leaderboard Endpoint\n    \n    Query Parameters:\n    - position_type: Filter by position type (long, short, both) - default: both\n    - limit: Number of models to return (max 100) - default: 50\n    - min_sample_size: Minimum sample size for inclusion - default: 10\n    - lookback_days: Days to look back for metrics - default: 30\n    - include_trends: Include trend indicators - default: true\n    \n    Returns:\n    - Ranked list of PTM models with performance metrics and trends\n    \"\"\"\n    return ptm_controller.get_leaderboard_endpoint()\n\n# Additional utility endpoints\n@app.route('/api/v1/ptm/leaderboard/summary', methods=['GET'])\ndef get_leaderboard_summary():\n    \"\"\"\n    Get leaderboard summary statistics\n    \"\"\"\n    try:\n        position_type_str = request.args.get('position_type', 'both').lower()\n        lookback_days = int(request.args.get('lookback_days', 30))\n        \n        # Get position type\n        position_type = None\n        if position_type_str in [e.value for e in PositionType]:\n            position_type = PositionType(position_type_str)\n        \n        cutoff_date = datetime.utcnow() - timedelta(days=lookback_days)\n        \n        # Query summary statistics\n        query = db.session.query(\n            func.count(PTMModel.id).label('total_models'),\n            func.avg(PTMPerformanceMetric.trust_score).label('avg_trust_score'),\n            func.avg(PTMPerformanceMetric.win_rate).label('avg_win_rate'),\n            func.avg(PTMPerformanceMetric.avg_return).label('avg_return'),\n            func.max(PTMPerformanceMetric.trust_score).label('max_trust_score'),\n            func.min(PTMPerformanceMetric.trust_score).label('min_trust_score')\n        ).select_from(PTMModel).join(PTMPerformanceMetric).filter(\n            PTMModel.is_active == True,\n            PTMPerformanceMetric.date >= cutoff_date.date()\n        )\n        \n        if position_type and position_type != PositionType.BOTH:\n            query = query.filter(PTMModel.position_type == position_type)\n        \n        result = query.first()\n        \n        return jsonify({\n            'success': True,\n            'data': {\n                'summary': {\n                    'total_active_models': result.total_models or 0,\n                    'average_trust_score': round(result.avg_trust_score or 0, 4),\n                    'average_win_rate': round(result.avg_win_rate or 0, 4),\n                    'average_return': round(result.avg_return or 0, 4),\n                    'top_trust_score': round(result.max_trust_score or 0, 4),\n                    'lowest_trust_score': round(result.min_trust_score or 0, 4)\n                },\n                'metadata': {\n                    'position_type_filter': position_type.value if position_type else 'all',\n                    'lookback_days': lookback_days,\n                    'generated_at': datetime.utcnow().isoformat()\n                }\n            }\n        }), 200\n        \n    except Exception as e:\n        return jsonify({\n            'success': False,\n            'error': {\n                'code': 'INTERNAL_ERROR',\n                'message': 'Failed to retrieve summary data'\n            }\n        }), 500\n\nif __name__ == '__main__':\n    app.run(debug=True)\n```\n\nThis implementation provides:\n\n## Key Features:\n\n1. **Comprehensive Ranking System**:\n   - Primary ranking by trust score\n   - Secondary ranking by average return\n   - Configurable minimum sample sizes\n\n2. **Rich Performance Metrics**:\n   - Trust scores with precision\n   - Win rates and average returns\n   - Sharpe ratios and max drawdown\n   - Sample sizes and trade counts\n\n3. **Trend Indicators**:\n   - 7-day trend analysis by default\n   - Direction indicators (up/down/stable)\n   - Magnitude of change\n   - Visual emoji indicators\n\n4. **Flexible Filtering**:\n   - Filter by position type (long/short/both)\n   - Configurable lookback periods\n   - Minimum sample size requirements\n   - Result limits with maximum cap\n\n5. **Additional Endpoints**:\n   - Summary statistics endpoint\n   - Metadata about filtering criteria\n   - Last updated timestamps\n\n## Usage Examples:\n\n```bash\n# Get top 20 models for long positions\nGET /api/v1/ptm/leaderboard?position_type=long&limit=20\n\n# Get models with high sample sizes\nGET /api/v1/ptm/leaderboard?min_sample_size=50&lookback_days=60\n\n# Get leaderboard without trends\nGET /api/v1/ptm/leaderboard?include_trends=false\n\n# Get summary statistics\nGET /api/v1/ptm/leaderboard/summary?position_type=short\n```\n\nThe endpoint returns comprehensive leaderboard data with performance metrics, trend analysis, and flexible filtering options for PTM model evaluation."
}