This is page 7 of 7. Use http://codebase.md/derekrbreese/fantasy-football-mcp-public?page={x} to view the full context.
# Directory Structure
```
├── .dockerignore
├── .env.example
├── .gitignore
├── config
│ ├── __init__.py
│ └── settings.py
├── Dockerfile
├── docs
│ ├── BYE_WEEKS_FIX.md
│ ├── LIVE_API_TEST_RESULTS.md
│ ├── LIVE_API_TESTING_SUMMARY.md
│ ├── PHASE_2B_REFACTOR_SUMMARY.md
│ ├── PROMPTS_AND_RESOURCES.md
│ ├── TEST_SUITE_SUMMARY.md
│ └── WAIVER_WIRE_VALIDATION_FIX.md
├── examples
│ ├── demo_enhancement_layer.py
│ ├── demos
│ │ ├── demo_consolidated_roster.py
│ │ ├── demo_ff_get_roster.py
│ │ └── demo_ff_get_waiver_wire.py
│ ├── example_client_llm_usage.py
│ └── hybrid_optimizer_example.py
├── fantasy_football_multi_league.py
├── fastmcp_server.py
├── INSTALLATION.md
├── LICENSE
├── lineup_optimizer.py
├── matchup_analyzer.py
├── position_normalizer.py
├── pyproject.toml
├── README.md
├── render.yaml
├── requirements.txt
├── sleeper_api.py
├── src
│ ├── __init__.py
│ ├── agents
│ │ ├── __init__.py
│ │ ├── cache_manager.py
│ │ ├── config.py
│ │ ├── data_fetcher.py
│ │ ├── decision.py
│ │ ├── draft_evaluator.py
│ │ ├── hybrid_optimizer.py
│ │ ├── integration.py
│ │ ├── llm_enhancement.py
│ │ ├── optimization.py
│ │ ├── reddit_analyzer.py
│ │ ├── roster_detector.py
│ │ ├── statistical.py
│ │ ├── user_interaction_engine.py
│ │ └── yahoo_auth.py
│ ├── api
│ │ ├── __init__.py
│ │ ├── yahoo_client.py
│ │ └── yahoo_utils.py
│ ├── data
│ │ └── bye_weeks_2025.json
│ ├── handlers
│ │ ├── __init__.py
│ │ ├── admin_handlers.py
│ │ ├── analytics_handlers.py
│ │ ├── draft_handlers.py
│ │ ├── league_handlers.py
│ │ ├── matchup_handlers.py
│ │ ├── player_handlers.py
│ │ └── roster_handlers.py
│ ├── lineup_optimizer.py
│ ├── matchup_analyzer.py
│ ├── mcp_server.py
│ ├── models
│ │ ├── __init__.py
│ │ ├── draft.py
│ │ ├── lineup.py
│ │ ├── matchup.py
│ │ └── player.py
│ ├── parsers
│ │ ├── __init__.py
│ │ └── yahoo_parsers.py
│ ├── position_normalizer.py
│ ├── services
│ │ ├── __init__.py
│ │ ├── player_enhancement.py
│ │ └── reddit_service.py
│ ├── sleeper_api.py
│ ├── strategies
│ │ ├── __init__.py
│ │ ├── aggressive.py
│ │ ├── balanced.py
│ │ ├── base.py
│ │ └── conservative.py
│ ├── utils
│ │ ├── __init__.py
│ │ ├── bye_weeks.py
│ │ ├── constants.py
│ │ ├── roster_configs.py
│ │ └── scoring.py
│ └── yahoo_api_utils.py
├── tests
│ ├── conftest.py
│ ├── integration
│ │ ├── __init__.py
│ │ └── test_mcp_tools.py
│ ├── README.md
│ ├── TEST_CLEANUP_PLAN.md
│ ├── test_enhancement_layer.py
│ ├── test_live_api.py
│ ├── test_real_data.py
│ └── unit
│ ├── __init__.py
│ ├── test_api_client.py
│ ├── test_bye_weeks_utility.py
│ ├── test_bye_weeks.py
│ ├── test_handlers.py
│ ├── test_lineup_optimizer.py
│ └── test_parsers.py
├── utils
│ ├── reauth_yahoo.py
│ ├── refresh_yahoo_token.py
│ ├── setup_yahoo_auth.py
│ └── verify_setup.py
└── yahoo_api_utils.py
```
# Files
--------------------------------------------------------------------------------
/fantasy_football_multi_league.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Fantasy Football MCP Server - Multi-League Support
"""
import asyncio
import json
import os
from typing import Any, Awaitable, Callable, Dict, List, Optional
from dotenv import load_dotenv
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import TextContent, Tool
# Import extracted modules
from src.api import get_access_token, refresh_yahoo_token, set_access_token, yahoo_api_call
from src.parsers import parse_team_roster, parse_yahoo_free_agent_players
from src.services import analyze_reddit_sentiment
# Import rate limiting and caching utilities
from src.api.yahoo_utils import rate_limiter, response_cache
# Import bye week utilities
from src.utils.bye_weeks import get_bye_week_with_fallback
# Import all handlers from the handlers module
from src.handlers import (
handle_ff_analyze_draft_state,
handle_ff_analyze_reddit_sentiment,
handle_ff_build_lineup,
handle_ff_clear_cache,
handle_ff_compare_teams,
handle_ff_get_api_status,
handle_ff_get_draft_rankings,
handle_ff_get_draft_recommendation,
handle_ff_get_draft_results,
handle_ff_get_league_info,
handle_ff_get_leagues,
handle_ff_get_matchup,
handle_ff_get_players,
handle_ff_get_roster,
handle_ff_get_standings,
handle_ff_get_teams,
handle_ff_get_waiver_wire,
handle_ff_refresh_token,
inject_draft_dependencies,
inject_league_helpers,
inject_matchup_dependencies,
inject_player_dependencies,
inject_roster_dependencies,
)
# Draft functionality is built-in (no complex imports needed)
DRAFT_AVAILABLE = True
# Load environment
load_dotenv()
# Initialize access token in the API module
if os.getenv("YAHOO_ACCESS_TOKEN"):
set_access_token(os.getenv("YAHOO_ACCESS_TOKEN"))
# Create server instance
server = Server("fantasy-football")
# Cache for leagues
LEAGUES_CACHE = {}
async def discover_leagues() -> dict[str, dict[str, Any]]:
"""Discover all active NFL leagues for the authenticated user."""
global LEAGUES_CACHE
if LEAGUES_CACHE:
return LEAGUES_CACHE
# Get current NFL leagues (game key 461 for 2025)
data = await yahoo_api_call("users;use_login=1/games;game_keys=nfl/leagues")
leagues = {}
try:
users = data.get("fantasy_content", {}).get("users", {})
if "0" in users:
user = users["0"]["user"]
if isinstance(user, list):
for item in user:
if isinstance(item, dict) and "games" in item:
games = item["games"]
if "0" in games: # First game (NFL)
game = games["0"]["game"]
if isinstance(game, list):
for g in game:
if isinstance(g, dict) and "leagues" in g:
league_data = g["leagues"]
for key in league_data:
if key != "count" and isinstance(
league_data[key], dict
):
if "league" in league_data[key]:
league_info = league_data[key]["league"]
if (
isinstance(league_info, list)
and len(league_info) > 0
):
league_dict = league_info[0]
league_key = league_dict.get(
"league_key", ""
)
leagues[league_key] = {
"key": league_key,
"id": league_dict.get("league_id", ""),
"name": league_dict.get(
"name", "Unknown"
),
"season": league_dict.get(
"season", 2025
),
"num_teams": league_dict.get(
"num_teams", 0
),
"scoring_type": league_dict.get(
"scoring_type", "head"
),
"current_week": league_dict.get(
"current_week", 1
),
"is_finished": league_dict.get(
"is_finished", 0
),
}
except Exception:
pass # Silently handle error to not interfere with MCP protocol
LEAGUES_CACHE = leagues
return leagues
async def get_user_team_info(league_key: Optional[str]) -> Optional[dict]:
if not league_key:
return None
"""Get the user's team details in a league.
Normalizes manager entries and `is_owned_by_current_login` flags so the
caller can reliably identify which team belongs to the authenticated user.
"""
try:
data = await yahoo_api_call(f"league/{league_key}/teams")
# Get user's GUID from environment
user_guid = os.getenv("YAHOO_GUID", "your_yahoo_guid_here")
# Parse to find user's team
league = data.get("fantasy_content", {}).get("league", [])
if len(league) > 1 and isinstance(league[1], dict) and "teams" in league[1]:
teams = league[1]["teams"]
for key in teams:
if key != "count" and isinstance(teams[key], dict):
if "team" in teams[key]:
team_array = teams[key]["team"]
if isinstance(team_array, list) and len(team_array) > 0:
# The team data is in the first element
team_data = team_array[0]
if isinstance(team_data, list):
team_key = None
team_name = None
is_users_team = False
draft_grade = None
draft_position = None
# Parse each element in the team data
for element in team_data:
if isinstance(element, dict):
# Check for team key
if "team_key" in element:
team_key = element["team_key"]
# Get team name
if "name" in element:
team_name = element["name"]
# Get draft grade
if "draft_grade" in element:
draft_grade = element["draft_grade"]
# Get draft position
if "draft_position" in element:
draft_position = element["draft_position"]
# Check if owned by current login (API may return int, bool or string)
owned_flag = element.get("is_owned_by_current_login")
if str(owned_flag) == "1" or owned_flag is True:
is_users_team = True
# Also check by GUID
if "managers" in element:
managers = element["managers"]
if isinstance(managers, dict):
managers = [
m
for key, m in managers.items()
if key != "count"
]
if managers:
mgr = managers[0].get("manager", {})
if mgr.get("guid") == user_guid:
is_users_team = True
if is_users_team and team_key:
return {
"team_key": team_key,
"team_name": team_name,
"draft_grade": draft_grade,
"draft_position": draft_position,
}
return None
except Exception:
# Silently handle error to not interfere with MCP protocol
return None
async def get_user_team_key(league_key: Optional[str]) -> Optional[str]:
if not league_key:
return None
"""Get the user's team key in a specific league (legacy function for compatibility)."""
team_info = await get_user_team_info(league_key)
return team_info["team_key"] if team_info else None
async def get_waiver_wire_players(
league_key: str, position: str = "all", sort: str = "rank", count: int = 30
) -> list[dict]:
"""Get available waiver wire players with detailed stats."""
try:
# Build the API call with filters
pos_filter = f";position={position}" if position != "all" else ""
sort_type = {
"rank": "OR", # Overall rank
"points": "PTS", # Points
"owned": "O", # Ownership %
"trending": "A", # Added %
}.get(sort, "OR")
endpoint = (
f"league/{league_key}/players;status=A{pos_filter};sort={sort_type};count={count}"
)
data = await yahoo_api_call(endpoint)
players = []
league = data.get("fantasy_content", {}).get("league", [])
# Players are in the second element of the league array
if len(league) > 1 and isinstance(league[1], dict) and "players" in league[1]:
players_data = league[1]["players"]
for key in players_data:
if key != "count" and isinstance(players_data[key], dict):
if "player" in players_data[key]:
player_array = players_data[key]["player"]
# Player data is in nested array structure
if isinstance(player_array, list) and len(player_array) > 0:
player_data = player_array[0]
if isinstance(player_data, list):
player_info = {}
for element in player_data:
if isinstance(element, dict):
# Basic info
if "name" in element:
player_info["name"] = element["name"]["full"]
if "player_key" in element:
player_info["player_key"] = element["player_key"]
if "editorial_team_abbr" in element:
player_info["team"] = element["editorial_team_abbr"]
if "display_position" in element:
player_info["position"] = element["display_position"]
# Extract bye week with fallback to static data
api_bye_week = None
if "bye_weeks" in element:
bye_weeks_data = element["bye_weeks"]
if isinstance(bye_weeks_data, dict) and "week" in bye_weeks_data:
bye_week = bye_weeks_data.get("week")
# Validate bye week is a valid week number (1-18)
if bye_week and str(bye_week).isdigit():
bye_num = int(bye_week)
if 1 <= bye_num <= 18:
api_bye_week = bye_num
# Use fallback utility to get bye week (tries API first, then static data)
team_abbr = element.get("editorial_team_abbr", "")
player_info["bye"] = get_bye_week_with_fallback(team_abbr, api_bye_week)
# Ownership data
if "ownership" in element:
ownership = element["ownership"]
player_info["owned_pct"] = ownership.get(
"ownership_percentage", 0
)
player_info["weekly_change"] = ownership.get(
"weekly_change", 0
)
# Injury status
if "status" in element:
player_info["injury_status"] = element["status"]
if "status_full" in element:
player_info["injury_detail"] = element["status_full"]
if player_info.get("name"):
# Ensure all expected fields are present with defaults
player_info.setdefault("team", "FA") # Free Agent if no team
player_info.setdefault(
"owned_pct", 0
) # 0% if no ownership data
player_info.setdefault(
"weekly_change", 0
) # No change if no data
player_info.setdefault(
"injury_status", "Healthy"
) # Assume healthy if not specified
players.append(player_info)
return players
except Exception:
return []
async def get_draft_rankings(
league_key: Optional[str] = None, position: str = "all", count: int = 50
) -> list[dict]:
"""Get pre-draft rankings with ADP data."""
try:
# If no league key provided, get the first available league
if not league_key:
leagues = await discover_leagues()
if leagues:
league_key = list(leagues.keys())[0]
else:
return [] # No leagues available
pos_filter = f";position={position}" if position != "all" else ""
# Get all players sorted by rank for the specified league
endpoint = f"league/{league_key}/players{pos_filter};sort=OR;count={count}"
data = await yahoo_api_call(endpoint)
players = []
league = data.get("fantasy_content", {}).get("league", [])
# Players are in the second element of the league array
if len(league) > 1 and isinstance(league[1], dict) and "players" in league[1]:
players_data = league[1]["players"]
for key in players_data:
if key != "count" and isinstance(players_data[key], dict):
if "player" in players_data[key]:
player_array = players_data[key]["player"]
# Player data is in nested array structure
if isinstance(player_array, list) and len(player_array) > 0:
player_data = player_array[0]
if isinstance(player_data, list):
player_info = {}
rank = int(key) + 1 # Use the key as rank
for element in player_data:
if isinstance(element, dict):
if "name" in element:
player_info["name"] = element["name"]["full"]
if "editorial_team_abbr" in element:
player_info["team"] = element["editorial_team_abbr"]
if "display_position" in element:
player_info["position"] = element["display_position"]
# Extract bye week with fallback to static data
api_bye_week = None
if "bye_weeks" in element:
bye_weeks_data = element["bye_weeks"]
if isinstance(bye_weeks_data, dict) and "week" in bye_weeks_data:
bye_week = bye_weeks_data.get("week")
# Validate bye week is a valid week number (1-18)
if bye_week and str(bye_week).isdigit():
bye_num = int(bye_week)
if 1 <= bye_num <= 18:
api_bye_week = bye_num
# Use fallback utility to get bye week (tries API first, then static data)
team_abbr = element.get("editorial_team_abbr", "")
player_info["bye"] = get_bye_week_with_fallback(team_abbr, api_bye_week)
# Draft data if available
if "draft_analysis" in element:
draft = element["draft_analysis"]
player_info["average_draft_position"] = draft.get(
"average_pick", rank
)
player_info["average_round"] = draft.get(
"average_round", "N/A"
)
player_info["average_cost"] = draft.get(
"average_cost", "N/A"
)
player_info["percent_drafted"] = draft.get(
"percent_drafted", 0
)
else:
# Use rank as ADP if no draft data
player_info["rank"] = rank
if player_info.get("name"):
players.append(player_info)
# Sort by ADP if available
players.sort(
key=lambda x: (
float(x.get("average_draft_position", 999))
if x.get("average_draft_position") != "N/A"
else 999
)
)
return players
except Exception:
return []
async def get_all_teams_info(league_key: str) -> list[dict]:
"""Get all teams information including draft data."""
try:
data = await yahoo_api_call(f"league/{league_key}/teams")
teams_list = []
league = data.get("fantasy_content", {}).get("league", [])
if len(league) > 1 and isinstance(league[1], dict) and "teams" in league[1]:
teams = league[1]["teams"]
for key in teams:
if key != "count" and isinstance(teams[key], dict):
if "team" in teams[key]:
team_array = teams[key]["team"]
if isinstance(team_array, list) and len(team_array) > 0:
team_data = team_array[0]
if isinstance(team_data, list):
team_info = {}
for element in team_data:
if isinstance(element, dict):
if "team_key" in element:
team_info["team_key"] = element["team_key"]
if "team_id" in element:
team_info["team_id"] = element["team_id"]
if "name" in element:
team_info["name"] = element["name"]
if "draft_grade" in element:
team_info["draft_grade"] = element["draft_grade"]
if "draft_position" in element:
team_info["draft_position"] = element["draft_position"]
if "draft_recap_url" in element:
team_info["draft_recap_url"] = element[
"draft_recap_url"
]
if "number_of_moves" in element:
team_info["moves"] = element["number_of_moves"]
if "number_of_trades" in element:
team_info["trades"] = element["number_of_trades"]
if "managers" in element:
managers = element["managers"]
if managers and len(managers) > 0:
mgr = managers[0].get("manager", {})
team_info["manager"] = mgr.get(
"nickname", "Unknown"
)
if team_info.get("team_key"):
teams_list.append(team_info)
# Sort by draft position if available
teams_list.sort(key=lambda x: x.get("draft_position", 999))
return teams_list
except Exception:
return []
@server.list_tools()
async def list_tools() -> list[Tool]:
"""List available fantasy football tools."""
base_tools = [
Tool(
name="ff_get_leagues",
description="Get all your fantasy football leagues",
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="ff_get_league_info",
description="Get detailed information about a specific league",
inputSchema={
"type": "object",
"properties": {
"league_key": {
"type": "string",
"description": "League key (e.g., 'nfl.l.XXXXXX'). Use ff_get_leagues to get available keys.",
}
},
"required": ["league_key"],
},
),
Tool(
name="ff_get_standings",
description="Get standings for a specific league",
inputSchema={
"type": "object",
"properties": {
"league_key": {
"type": "string",
"description": "League key (e.g., 'nfl.l.XXXXXX')",
}
},
"required": ["league_key"],
},
),
Tool(
name="ff_get_teams",
description="Get all teams in a specific league with basic information",
inputSchema={
"type": "object",
"properties": {
"league_key": {
"type": "string",
"description": "League key (e.g., 'nfl.l.XXXXXX')",
}
},
"required": ["league_key"],
},
),
Tool(
name="ff_get_roster",
description="Get your team roster in a specific league",
inputSchema={
"type": "object",
"properties": {
"league_key": {
"type": "string",
"description": "League key (e.g., 'nfl.l.XXXXXX')",
},
"team_key": {
"anyOf": [{"type": "string"}, {"type": "null"}],
"description": "Optional team key if not the logged-in team",
},
"week": {
"anyOf": [{"type": "integer"}, {"type": "null"}],
"description": "Week for projections and analysis (optional, defaults to current)",
},
"data_level": {
"type": "string",
"description": "Data detail level: 'basic', 'standard', 'enhanced'",
"enum": ["basic", "standard", "enhanced"],
"default": "standard",
},
"include_analysis": {
"anyOf": [{"type": "boolean"}, {"type": "null"}],
"description": "Include basic roster analysis",
"default": False,
},
"include_projections": {
"anyOf": [{"type": "boolean"}, {"type": "null"}],
"description": "Include projections from Yahoo and Sleeper",
"default": True,
},
"include_external_data": {
"anyOf": [{"type": "boolean"}, {"type": "null"}],
"description": "Include Sleeper data, trending, and matchups",
"default": True,
},
},
"required": ["league_key"],
},
),
Tool(
name="ff_get_matchup",
description="Get matchup for a specific week in a league",
inputSchema={
"type": "object",
"properties": {
"league_key": {
"type": "string",
"description": "League key (e.g., 'nfl.l.XXXXXX')",
},
"week": {
"type": "integer",
"description": "Week number (optional, defaults to current week)",
},
},
"required": ["league_key"],
},
),
Tool(
name="ff_get_players",
description="Get available free agent players in a league",
inputSchema={
"type": "object",
"properties": {
"league_key": {
"type": "string",
"description": "League key (e.g., 'nfl.l.XXXXXX')",
},
"position": {
"type": "string",
"description": "Position filter (QB, RB, WR, TE, K, DEF)",
},
"count": {
"type": "integer",
"description": "Number of players to return",
"default": 10,
},
"sort": {
"type": "string",
"description": "Sort by: 'rank', 'points', 'owned', 'trending'",
"enum": ["rank", "points", "owned", "trending"],
"default": "rank",
},
"week": {
"anyOf": [{"type": "integer"}, {"type": "null"}],
"description": "Week for projections and analysis (optional, defaults to current)",
},
"include_analysis": {
"anyOf": [{"type": "boolean"}, {"type": "null"}],
"description": "Include basic analysis and rankings",
"default": False,
},
"include_expert_analysis": {
"anyOf": [{"type": "boolean"}, {"type": "null"}],
"description": "Include expert analysis and recommendations",
"default": False,
},
"include_projections": {
"anyOf": [{"type": "boolean"}, {"type": "null"}],
"description": "Include projections from Yahoo and Sleeper",
"default": True,
},
"include_external_data": {
"anyOf": [{"type": "boolean"}, {"type": "null"}],
"description": "Include Sleeper data, trending, and matchups",
"default": True,
},
},
"required": ["league_key"],
},
),
Tool(
name="ff_compare_teams",
description="Compare two teams' rosters within a league",
inputSchema={
"type": "object",
"properties": {
"league_key": {
"type": "string",
"description": "League key (e.g., 'nfl.l.XXXXXX')",
},
"team_key_a": {
"type": "string",
"description": "First team key to compare",
},
"team_key_b": {
"type": "string",
"description": "Second team key to compare",
},
},
"required": ["league_key", "team_key_a", "team_key_b"],
},
),
Tool(
name="ff_build_lineup",
description="Build optimal lineup from your roster using strategy-based optimization and positional constraints",
inputSchema={
"type": "object",
"properties": {
"league_key": {
"type": "string",
"description": "League key (e.g., 'nfl.l.XXXXXX')",
},
"week": {
"type": "integer",
"description": "Week number (optional, defaults to current week)",
},
"strategy": {
"type": "string",
"description": "Strategy: 'conservative', 'aggressive', or 'balanced' (default: balanced)",
"enum": ["conservative", "aggressive", "balanced"],
},
"use_llm": {
"type": "boolean",
"description": "Use LLM-based optimization instead of mathematical formulas (default: false)",
},
},
"required": ["league_key"],
},
),
Tool(
name="ff_refresh_token",
description="Refresh the Yahoo API access token when it expires",
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="ff_get_draft_results",
description="Get draft results showing all teams with their draft positions and grades",
inputSchema={
"type": "object",
"properties": {
"league_key": {
"type": "string",
"description": "League key (e.g., 'nfl.l.XXXXXX')",
},
"team_key": {
"type": "string",
"description": "Optional team key if not the logged-in team",
},
},
"required": ["league_key"],
},
),
Tool(
name="ff_get_waiver_wire",
description="Get top available waiver wire players with detailed stats and projections",
inputSchema={
"type": "object",
"properties": {
"league_key": {
"type": "string",
"description": "League key (e.g., 'nfl.l.XXXXXX')",
},
"position": {
"type": "string",
"description": "Position filter (QB, RB, WR, TE, K, DEF, or 'all')",
"enum": ["QB", "RB", "WR", "TE", "K", "DEF", "all"],
},
"sort": {
"type": "string",
"description": "Sort by: 'rank', 'points', 'owned', 'trending'",
"enum": ["rank", "points", "owned", "trending"],
"default": "rank",
},
"count": {
"type": "integer",
"description": "Number of players to return (default: 30)",
"default": 30,
},
"week": {
"anyOf": [{"type": "integer"}, {"type": "null"}],
"description": "Week for projections and analysis (optional, defaults to current)",
},
"team_key": {
"anyOf": [{"type": "string"}, {"type": "null"}],
"description": "Optional team key for context (e.g., waiver priority)",
},
"include_analysis": {
"anyOf": [{"type": "boolean"}, {"type": "null"}],
"description": "Include basic waiver priority analysis",
"default": False,
},
"include_projections": {
"anyOf": [{"type": "boolean"}, {"type": "null"}],
"description": "Include projections from Yahoo and Sleeper",
"default": True,
},
"include_external_data": {
"anyOf": [{"type": "boolean"}, {"type": "null"}],
"description": "Include Sleeper data, trending, and matchups",
"default": True,
},
},
"required": ["league_key"],
},
),
Tool(
name="ff_get_api_status",
description="Get Yahoo API rate limit status and cache statistics",
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="ff_clear_cache",
description="Clear the API response cache",
inputSchema={
"type": "object",
"properties": {
"pattern": {
"type": "string",
"description": "Optional pattern to match (e.g., 'standings', 'roster'). Clears all if not provided.",
}
},
},
),
Tool(
name="ff_get_draft_rankings",
description="Get pre-draft player rankings and ADP (Average Draft Position)",
inputSchema={
"type": "object",
"properties": {
"league_key": {
"type": "string",
"description": "League key (optional, uses first available league if not provided)",
},
"position": {
"type": "string",
"description": "Position filter (QB, RB, WR, TE, K, DEF, or 'all')",
"enum": ["QB", "RB", "WR", "TE", "K", "DEF", "all"],
"default": "all",
},
"count": {
"type": "integer",
"description": "Number of players to return (default: 50)",
"default": 50,
},
},
"required": [],
},
),
]
# Add draft tools if available
if DRAFT_AVAILABLE:
draft_tools = [
Tool(
name="ff_get_draft_recommendation",
description="Get AI-powered draft recommendations for live fantasy football drafts",
inputSchema={
"type": "object",
"properties": {
"league_key": {
"type": "string",
"description": "League key (e.g., 'nfl.l.XXXXXX')",
},
"strategy": {
"type": "string",
"description": "Draft strategy: 'conservative', 'aggressive', or 'balanced' (default: balanced)",
"enum": ["conservative", "aggressive", "balanced"],
"default": "balanced",
},
"num_recommendations": {
"type": "integer",
"description": "Number of top recommendations to return (1-20, default: 10)",
"minimum": 1,
"maximum": 20,
"default": 10,
},
"current_pick": {
"type": "integer",
"description": "Current overall pick number (optional)",
},
},
"required": ["league_key"],
},
),
Tool(
name="ff_analyze_draft_state",
description="Analyze current draft state including roster needs and strategic insights",
inputSchema={
"type": "object",
"properties": {
"league_key": {
"type": "string",
"description": "League key (e.g., 'nfl.l.XXXXXX')",
},
"strategy": {
"type": "string",
"description": "Draft strategy for analysis: 'conservative', 'aggressive', or 'balanced' (default: balanced)",
"enum": ["conservative", "aggressive", "balanced"],
"default": "balanced",
},
},
"required": ["league_key"],
},
),
Tool(
name="ff_analyze_reddit_sentiment",
description="Analyze Reddit sentiment for fantasy football players to help with Start/Sit decisions",
inputSchema={
"type": "object",
"properties": {
"players": {
"type": "array",
"items": {"type": "string"},
"description": "List of player names to analyze (e.g., ['Josh Allen', 'Jared Goff'])",
},
"time_window_hours": {
"type": "integer",
"description": "How far back to look for Reddit posts (default: 48 hours)",
"default": 48,
},
},
"required": ["players"],
},
),
]
return base_tools + draft_tools
return base_tools
TOOL_HANDLERS: dict[str, Callable[[dict], Awaitable[dict]]] = {
"ff_get_leagues": handle_ff_get_leagues,
"ff_get_league_info": handle_ff_get_league_info,
"ff_get_standings": handle_ff_get_standings,
"ff_get_teams": handle_ff_get_teams,
"ff_get_roster": handle_ff_get_roster,
"ff_get_roster_with_projections": handle_ff_get_roster,
"ff_get_matchup": handle_ff_get_matchup,
"ff_get_players": handle_ff_get_players,
"ff_compare_teams": handle_ff_compare_teams,
"ff_build_lineup": handle_ff_build_lineup,
"ff_refresh_token": handle_ff_refresh_token,
"ff_get_api_status": handle_ff_get_api_status,
"ff_clear_cache": handle_ff_clear_cache,
"ff_get_draft_results": handle_ff_get_draft_results,
"ff_get_waiver_wire": handle_ff_get_waiver_wire,
"ff_get_draft_rankings": handle_ff_get_draft_rankings,
"ff_get_draft_recommendation": handle_ff_get_draft_recommendation,
"ff_analyze_draft_state": handle_ff_analyze_draft_state,
"ff_analyze_reddit_sentiment": handle_ff_analyze_reddit_sentiment,
}
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
"""Execute a fantasy football tool via modular handlers."""
original_arguments = dict(arguments)
handler_args = {k: v for k, v in original_arguments.items() if k != "debug"}
debug_flag = original_arguments.get("debug") is True
debug_msgs: list[str] = []
if debug_flag:
debug_msgs.append(f"debug: call_tool entered for {name}")
try:
handler = TOOL_HANDLERS.get(name)
if handler is None:
result: Any = {"error": f"Unknown tool: {name}"}
else:
result = await handler(handler_args)
if isinstance(result, str) and result.strip() == "0":
result = {
"status": "error",
"message": "Internal legacy layer produced sentinel '0' string",
"tool": name,
"stage": "legacy.call_tool.guard",
}
# Ensure result is always a dict for consistent handling
if isinstance(result, str):
result = {"content": result}
if debug_flag:
safe_args = {
key: value
for key, value in handler_args.items()
if not key.lower().endswith("token")
}
debug_msgs.append(f"debug: sanitized arguments -> {sorted(safe_args.keys())}")
result["_debug"] = {
"messages": debug_msgs,
"tool": name,
"arguments": safe_args,
}
return [TextContent(type="text", text=json.dumps(result, indent=2))]
except Exception as exc: # pragma: no cover - defensive catch
error_result = {
"error": str(exc),
"tool": name,
"arguments": original_arguments,
}
return [TextContent(type="text", text=json.dumps(error_result, indent=2))]
async def get_draft_recommendation_simple(
league_key: str, strategy: str, num_recommendations: int, current_pick: Optional[int] = None
) -> dict:
"""Simplified draft recommendation using available data."""
try:
# Get available players using existing waiver wire function
available_players = await get_waiver_wire_players(league_key, count=100)
draft_rankings = await get_draft_rankings(league_key, count=50)
# Simple scoring based on rankings and availability
recommendations = []
# Create a quick lookup for available players
available_names = {p.get("name", "").lower() for p in available_players}
for player in draft_rankings:
player_name = player.get("name", "").lower()
if player_name in available_names:
# Simple scoring based on strategy
rank = player.get("rank", 999)
base_score = max(0, 100 - rank)
if strategy == "conservative":
# Prefer higher-ranked (safer) picks
score = base_score + (10 if rank <= 24 else 0)
reasoning = f"Rank #{rank}, conservative choice (proven player)"
elif strategy == "aggressive":
# Prefer potential breakouts (lower owned %)
owned_pct = next(
(
p.get("owned_pct", 50)
for p in available_players
if p.get("name", "").lower() == player_name
),
50,
)
upside_bonus = max(0, 20 - (owned_pct / 5)) # Bonus for lower ownership
score = base_score + upside_bonus
reasoning = f"Rank #{rank}, high upside potential ({owned_pct}% owned)"
else: # balanced
score = base_score + (5 if rank <= 50 else 0)
reasoning = f"Rank #{rank}, balanced value pick"
recommendations.append({"player": player, "score": score, "reasoning": reasoning})
# Sort by score and take top N
recommendations.sort(key=lambda x: x["score"], reverse=True)
top_picks = recommendations[:num_recommendations]
return {
"status": "success",
"league_key": league_key,
"strategy": strategy,
"current_pick": current_pick,
"recommendations": top_picks,
"total_analyzed": len(recommendations),
"insights": [
f"Using {strategy} draft strategy",
f"Analyzed {len(available_players)} available players",
"Cross-referenced with Yahoo rankings",
"Recommendations prioritize available players only",
],
}
except Exception as e:
return {
"status": "error",
"error": f"Draft recommendation failed: {str(e)}",
"fallback": "Use ff_get_draft_rankings and ff_get_players for manual analysis",
}
async def analyze_draft_state_simple(league_key: str, strategy: str) -> dict:
"""Simplified draft state analysis."""
try:
# Get current roster and league info
await yahoo_api_call(f"league/{league_key}/teams")
leagues = await discover_leagues()
league_info = leagues.get(league_key, {})
# Analyze positional needs (simplified)
user_team = await get_user_team_info(league_key)
# Get current week to estimate draft progress
current_week = league_info.get("current_week", 1)
draft_phase = "pre_season" if current_week <= 1 else "mid_season"
positional_needs = {
"QB": "medium", # Usually need 1-2
"RB": "high", # Need 3-5
"WR": "high", # Need 3-5
"TE": "medium", # Need 1-2
"K": "low", # Stream position
"DEF": "low", # Stream position
}
strategic_advice = []
if strategy == "conservative":
strategic_advice.append("Focus on proven players with consistent production")
strategic_advice.append("Avoid injury-prone or rookie players early")
elif strategy == "aggressive":
strategic_advice.append("Target high-upside players and breakout candidates")
strategic_advice.append("Consider reaching for players with league-winning potential")
else:
strategic_advice.append("Balance safety with upside potential")
strategic_advice.append("Follow tier-based drafting approach")
return {
"status": "success",
"league_key": league_key,
"strategy": strategy,
"analysis": {
"draft_phase": draft_phase,
"league_info": {
"name": league_info.get("name", "Unknown"),
"teams": league_info.get("num_teams", 12),
"scoring": league_info.get("scoring_type", "standard"),
},
"positional_needs": positional_needs,
"strategic_advice": strategic_advice,
"your_team": (
user_team.get("team_name", "Unknown") if user_team else "Team info unavailable"
),
},
"recommendations": [
"Use ff_get_draft_recommendation for specific player suggestions",
"Monitor ff_get_players for available free agents",
"Check ff_get_draft_rankings for current ADP data",
],
}
except Exception as e:
return {
"status": "error",
"error": f"Draft analysis failed: {str(e)}",
"basic_info": "Use ff_get_league_info for basic league details",
}
# ==============================================================================
# DEPENDENCY INJECTION - Wire up handler dependencies
# ==============================================================================
# Inject dependencies for league handlers
inject_league_helpers(
discover_leagues=discover_leagues,
get_user_team_info=get_user_team_info,
get_all_teams_info=get_all_teams_info,
)
# Inject dependencies for roster handlers
inject_roster_dependencies(
get_user_team_info=get_user_team_info,
yahoo_api_call=yahoo_api_call,
parse_team_roster=parse_team_roster,
)
# Inject dependencies for matchup handlers
inject_matchup_dependencies(
get_user_team_key=get_user_team_key,
get_user_team_info=get_user_team_info,
yahoo_api_call=yahoo_api_call,
parse_team_roster=parse_team_roster,
)
# Inject dependencies for player handlers
inject_player_dependencies(
yahoo_api_call=yahoo_api_call,
get_waiver_wire_players=get_waiver_wire_players,
)
# Inject dependencies for draft handlers
inject_draft_dependencies(
get_all_teams_info=get_all_teams_info,
get_draft_rankings=get_draft_rankings,
get_draft_recommendation_simple=get_draft_recommendation_simple,
analyze_draft_state_simple=analyze_draft_state_simple,
DRAFT_AVAILABLE=DRAFT_AVAILABLE,
)
async def main():
"""Run the MCP server."""
# Use stdio transport
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, server.create_initialization_options())
if __name__ == "__main__":
asyncio.run(main())
```
--------------------------------------------------------------------------------
/src/agents/optimization.py:
--------------------------------------------------------------------------------
```python
"""
High-performance lineup optimization agent with parallel processing.
This module implements advanced lineup optimization strategies using:
- Massive parallel processing with asyncio and concurrent.futures
- Genetic algorithms for large solution spaces
- Smart pruning to reduce search space
- Correlation-based stacking strategies
- Multiple optimization objectives (points, value, ownership)
"""
import asyncio
import itertools
import logging
import random
import time
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
from dataclasses import dataclass
from decimal import Decimal
from enum import Enum
from typing import Any, Dict, List, Optional, Set, Tuple, Union
from functools import partial
import numpy as np
from pydantic import BaseModel, Field
from ..models.player import Player, Position, PlayerProjections
from ..models.lineup import (
Lineup,
LineupSlot,
LineupConstraints,
LineupRecommendation,
LineupAlternative,
OptimizationStrategy,
LineupType,
)
logger = logging.getLogger(__name__)
class OptimizationObjective(str, Enum):
"""Optimization objectives for lineup construction."""
MAXIMIZE_POINTS = "maximize_points"
MAXIMIZE_VALUE = "maximize_value"
MINIMIZE_OWNERSHIP = "minimize_ownership"
MAXIMIZE_CEILING = "maximize_ceiling"
MAXIMIZE_FLOOR = "maximize_floor"
BALANCED = "balanced"
@dataclass
class OptimizationWeights:
"""Weights for multi-objective optimization."""
points: float = 0.7
value: float = 0.2
ownership: float = 0.1
ceiling: float = 0.0
floor: float = 0.0
correlation: float = 0.0
variance_penalty: float = 0.0
@dataclass
class GeneticAlgorithmConfig:
"""Configuration for genetic algorithm optimization."""
population_size: int = 1000
generations: int = 200
mutation_rate: float = 0.1
crossover_rate: float = 0.8
elitism_rate: float = 0.1
tournament_size: int = 5
diversity_threshold: float = 0.8
class LineupChromosome:
"""Genetic algorithm chromosome representing a lineup."""
def __init__(self, players: List[Player], constraints: LineupConstraints):
self.players = players
self.constraints = constraints
self.fitness: Optional[float] = None
self.violations: List[str] = []
self._lineup: Optional[Lineup] = None
def to_lineup(self) -> Lineup:
"""Convert chromosome to Lineup model."""
if self._lineup is None:
slots = []
total_salary = 0
total_points = Decimal("0")
for i, player in enumerate(self.players):
position = self._get_position_for_slot(i)
salary = self._get_player_salary(player)
slot = LineupSlot(position=position, player=player, salary_used=salary)
slots.append(slot)
total_salary += salary
if player.projections:
total_points += player.projections.projected_fantasy_points
self._lineup = Lineup(
lineup_type=LineupType.DRAFTKINGS,
slots=slots,
total_salary=total_salary,
salary_remaining=self.constraints.salary_cap - total_salary,
salary_cap=self.constraints.salary_cap,
total_projected_points=total_points,
confidence_score=Decimal("0.8"),
)
return self._lineup
def _get_position_for_slot(self, slot_index: int) -> Position:
"""Map slot index to position requirement."""
# Standard DraftKings lineup: QB, RB, RB, WR, WR, WR, TE, FLEX, DST
position_map = {
0: Position.QB,
1: Position.RB,
2: Position.RB,
3: Position.WR,
4: Position.WR,
5: Position.WR,
6: Position.TE,
7: Position.RB, # FLEX - simplified to RB
8: Position.DEF,
}
return position_map.get(slot_index, Position.RB)
def _get_player_salary(self, player: Player) -> int:
"""Get player salary for optimization."""
if player.value_metrics:
return (
player.value_metrics.draftkings_salary
or player.value_metrics.fanduel_salary
or player.value_metrics.yahoo_salary
or 5000
)
return 5000
class OptimizationAgent:
"""High-performance lineup optimization agent with parallel processing."""
def __init__(self, max_workers: int = None):
"""Initialize the optimization agent.
Args:
max_workers: Maximum number of worker threads/processes
"""
self.max_workers = max_workers or min(32, (asyncio.get_event_loop().get_debug() and 1) or 4)
self.logger = logging.getLogger(__name__)
self._correlation_cache: Dict[str, Dict[str, float]] = {}
# Performance tracking
self.optimization_stats = {"total_evaluations": 0, "cache_hits": 0, "parallel_tasks": 0}
async def optimize_lineup(
self,
players: List[Player],
constraints: LineupConstraints,
strategy: OptimizationStrategy = OptimizationStrategy.BALANCED,
objective: OptimizationObjective = OptimizationObjective.MAXIMIZE_POINTS,
weights: Optional[OptimizationWeights] = None,
use_genetic_algorithm: bool = True,
max_alternatives: int = 5,
) -> LineupRecommendation:
"""Optimize lineup with parallel processing and multiple strategies.
Args:
players: Available players for selection
constraints: Lineup construction constraints
strategy: Optimization strategy
objective: Primary optimization objective
weights: Custom optimization weights
use_genetic_algorithm: Whether to use genetic algorithm for large spaces
max_alternatives: Maximum alternative lineups to generate
Returns:
LineupRecommendation with optimal lineup and alternatives
"""
start_time = time.time()
self.logger.info(f"Starting lineup optimization with {len(players)} players")
# Set default weights based on strategy
if weights is None:
weights = self._get_strategy_weights(strategy)
# Filter and validate players
valid_players = await self._filter_valid_players(players, constraints)
self.logger.info(f"Filtered to {len(valid_players)} valid players")
if len(valid_players) < 9: # Minimum for a lineup
raise ValueError("Insufficient players to form a valid lineup")
# Determine optimization approach
search_space_size = self._estimate_search_space(valid_players)
self.logger.info(f"Estimated search space size: {search_space_size:,}")
if use_genetic_algorithm and search_space_size > 10**6:
optimal_lineup = await self._genetic_algorithm_optimization(
valid_players, constraints, weights, objective
)
else:
optimal_lineup = await self._parallel_bruteforce_optimization(
valid_players, constraints, weights, objective
)
# Generate alternative lineups
alternatives = await self._generate_alternatives(
valid_players, constraints, weights, optimal_lineup, max_alternatives
)
# Create recommendation
recommendation = self._create_recommendation(
optimal_lineup, alternatives, strategy, weights
)
optimization_time = time.time() - start_time
self.logger.info(f"Optimization completed in {optimization_time:.2f} seconds")
return recommendation
async def rank_waiver_targets(
self,
available_players: List[Player],
current_roster: List[Player],
constraints: LineupConstraints,
weeks_ahead: int = 4,
) -> List[Tuple[Player, float, str]]:
"""Rank waiver wire targets based on lineup impact.
Args:
available_players: Players available on waivers
current_roster: Current roster players
constraints: Lineup constraints
weeks_ahead: Number of weeks to project
Returns:
List of (player, impact_score, reasoning) tuples
"""
self.logger.info(f"Ranking {len(available_players)} waiver targets")
# Create tasks for parallel evaluation
tasks = []
for player in available_players:
task = self._evaluate_waiver_impact(player, current_roster, constraints, weeks_ahead)
tasks.append(task)
# Execute in parallel with task groups
async with asyncio.TaskGroup() as tg:
results = [tg.create_task(task) for task in tasks]
# Collect and sort results
player_rankings = []
for i, result in enumerate(results):
impact_score, reasoning = await result
player_rankings.append((available_players[i], impact_score, reasoning))
# Sort by impact score descending
player_rankings.sort(key=lambda x: x[1], reverse=True)
return player_rankings[:20] # Top 20 targets
async def find_injury_replacements(
self,
injured_players: List[Player],
available_players: List[Player],
constraints: LineupConstraints,
max_replacements: int = 3,
) -> Dict[str, List[Tuple[Player, float, str]]]:
"""Find optimal injury replacements with parallel processing.
Args:
injured_players: Players who are injured
available_players: Available replacement players
constraints: Lineup constraints
max_replacements: Maximum replacements per injured player
Returns:
Dict mapping injured player ID to list of (replacement, score, reason)
"""
self.logger.info(f"Finding replacements for {len(injured_players)} injured players")
replacement_map = {}
# Create parallel tasks for each injured player
tasks = []
for injured_player in injured_players:
task = self._find_position_replacements(
injured_player, available_players, constraints, max_replacements
)
tasks.append((injured_player.id, task))
# Execute in parallel
results = await asyncio.gather(*[task for _, task in tasks])
# Map results back to injured players
for i, (player_id, _) in enumerate(tasks):
replacement_map[player_id] = results[i]
return replacement_map
async def _filter_valid_players(
self, players: List[Player], constraints: LineupConstraints
) -> List[Player]:
"""Filter players based on constraints with parallel validation."""
async def is_player_valid(player: Player) -> bool:
"""Check if player meets basic constraints."""
# Check exclusions
if constraints.excluded_players and player.id in constraints.excluded_players:
return False
# Check salary constraints
if player.value_metrics:
salary = (
player.value_metrics.draftkings_salary
or player.value_metrics.fanduel_salary
or player.value_metrics.yahoo_salary
)
if salary and salary > constraints.salary_cap:
return False
# Check injury status
if player.is_injured():
# Allow questionable players but not out/doubtful
injury_status = player.injury_report.status if player.injury_report else None
if injury_status in ["Out", "Doubtful", "IR"]:
return False
# Check projections
if not player.projections or not player.projections.projected_fantasy_points:
return False
return True
# Parallel validation
tasks = [is_player_valid(player) for player in players]
validity_results = await asyncio.gather(*tasks)
return [player for player, is_valid in zip(players, validity_results) if is_valid]
async def _parallel_bruteforce_optimization(
self,
players: List[Player],
constraints: LineupConstraints,
weights: OptimizationWeights,
objective: OptimizationObjective,
) -> Lineup:
"""Parallel brute force optimization with smart pruning."""
# Group players by position for efficient combination generation
players_by_position = self._group_players_by_position(players)
# Generate position combinations with pruning
position_combinations = await self._generate_position_combinations(
players_by_position, constraints
)
self.logger.info(f"Generated {len(position_combinations)} position combinations")
# Parallel evaluation of combinations
best_lineup = None
best_score = float("-inf")
# Process in batches to manage memory
batch_size = min(1000, len(position_combinations) // self.max_workers + 1)
for i in range(0, len(position_combinations), batch_size):
batch = position_combinations[i : i + batch_size]
# Create evaluation tasks
tasks = []
for combination in batch:
task = self._evaluate_lineup_combination(
combination, constraints, weights, objective
)
tasks.append(task)
# Execute batch in parallel
batch_results = await asyncio.gather(*tasks, return_exceptions=True)
# Process results
for result in batch_results:
if isinstance(result, Exception):
continue
lineup, score = result
if score > best_score:
best_score = score
best_lineup = lineup
if best_lineup is None:
raise ValueError("No valid lineup found with given constraints")
return best_lineup
async def _genetic_algorithm_optimization(
self,
players: List[Player],
constraints: LineupConstraints,
weights: OptimizationWeights,
objective: OptimizationObjective,
config: Optional[GeneticAlgorithmConfig] = None,
) -> Lineup:
"""Genetic algorithm optimization with parallel fitness evaluation."""
if config is None:
config = GeneticAlgorithmConfig()
self.logger.info(f"Starting genetic algorithm with population {config.population_size}")
# Initialize population
population = await self._initialize_population(players, constraints, config)
best_fitness = float("-inf")
best_chromosome = None
generations_without_improvement = 0
for generation in range(config.generations):
# Parallel fitness evaluation
await self._evaluate_population_fitness(population, weights, objective)
# Track best solution
current_best = max(population, key=lambda x: x.fitness or float("-inf"))
if current_best.fitness and current_best.fitness > best_fitness:
best_fitness = current_best.fitness
best_chromosome = current_best
generations_without_improvement = 0
self.logger.info(f"Generation {generation}: New best fitness {best_fitness:.3f}")
else:
generations_without_improvement += 1
# Early stopping
if generations_without_improvement > 50:
self.logger.info(f"Early stopping at generation {generation}")
break
# Create next generation
population = await self._create_next_generation(
population, players, constraints, config
)
if best_chromosome is None:
raise ValueError("Genetic algorithm failed to find valid solution")
return best_chromosome.to_lineup()
async def _generate_alternatives(
self,
players: List[Player],
constraints: LineupConstraints,
weights: OptimizationWeights,
optimal_lineup: Lineup,
max_alternatives: int,
) -> List[LineupAlternative]:
"""Generate alternative lineups with different strategies."""
alternatives = []
# Alternative strategies to try
alternative_objectives = [
(OptimizationObjective.MAXIMIZE_VALUE, "Value-focused alternative"),
(OptimizationObjective.MINIMIZE_OWNERSHIP, "Low-ownership contrarian play"),
(OptimizationObjective.MAXIMIZE_CEILING, "High-ceiling tournament play"),
(OptimizationObjective.MAXIMIZE_FLOOR, "Safe cash game play"),
]
# Generate alternatives in parallel
tasks = []
for objective, reason in alternative_objectives[:max_alternatives]:
task = self._create_alternative_lineup(
players, constraints, weights, objective, optimal_lineup, reason
)
tasks.append(task)
alternative_results = await asyncio.gather(*tasks, return_exceptions=True)
for result in alternative_results:
if isinstance(result, Exception):
continue
if result is not None:
alternatives.append(result)
return alternatives
async def _evaluate_waiver_impact(
self,
player: Player,
current_roster: List[Player],
constraints: LineupConstraints,
weeks_ahead: int,
) -> Tuple[float, str]:
"""Evaluate waiver player's impact on lineup optimization."""
# Create extended roster with waiver player
extended_roster = current_roster + [player]
# Optimize lineup with and without the player
try:
with_player_lineup = await self._quick_optimization(extended_roster, constraints)
without_player_lineup = await self._quick_optimization(current_roster, constraints)
# Calculate impact
point_improvement = (
with_player_lineup.total_projected_points
- without_player_lineup.total_projected_points
)
# Factor in upcoming matchups and consistency
upside_factor = self._calculate_upside_factor(player, weeks_ahead)
consistency_factor = self._calculate_consistency_factor(player)
impact_score = float(point_improvement) * upside_factor * consistency_factor
reasoning = f"Projected {point_improvement:.1f} point improvement, {upside_factor:.1f}x upside, {consistency_factor:.1f}x consistency"
return impact_score, reasoning
except Exception as e:
self.logger.warning(f"Error evaluating waiver impact for {player.name}: {e}")
return 0.0, "Evaluation error"
async def _find_position_replacements(
self,
injured_player: Player,
available_players: List[Player],
constraints: LineupConstraints,
max_replacements: int,
) -> List[Tuple[Player, float, str]]:
"""Find best replacements for an injured player."""
# Filter to same position
position_matches = [p for p in available_players if p.position == injured_player.position]
if not position_matches:
return []
# Evaluate replacements in parallel
tasks = []
for replacement in position_matches:
task = self._evaluate_replacement_player(injured_player, replacement, constraints)
tasks.append(task)
replacement_scores = await asyncio.gather(*tasks)
# Create ranked list
replacements = []
for i, (score, reason) in enumerate(replacement_scores):
replacements.append((position_matches[i], score, reason))
# Sort and return top replacements
replacements.sort(key=lambda x: x[1], reverse=True)
return replacements[:max_replacements]
def _group_players_by_position(self, players: List[Player]) -> Dict[Position, List[Player]]:
"""Group players by position for efficient combination generation."""
groups = {}
for player in players:
if player.position not in groups:
groups[player.position] = []
groups[player.position].append(player)
# Sort each position group by projected points descending
for position in groups:
groups[position].sort(
key=lambda p: (
p.projections.projected_fantasy_points if p.projections else Decimal("0")
),
reverse=True,
)
return groups
async def _generate_position_combinations(
self, players_by_position: Dict[Position, List[Player]], constraints: LineupConstraints
) -> List[List[Player]]:
"""Generate valid position combinations with smart pruning."""
# Standard DraftKings positions: QB(1), RB(2), WR(3), TE(1), DEF(1), FLEX(1)
position_requirements = {
Position.QB: 1,
Position.RB: 2,
Position.WR: 3,
Position.TE: 1,
Position.DEF: 1,
}
# Limit players per position for feasibility
max_players_per_position = {
Position.QB: min(5, len(players_by_position.get(Position.QB, []))),
Position.RB: min(10, len(players_by_position.get(Position.RB, []))),
Position.WR: min(15, len(players_by_position.get(Position.WR, []))),
Position.TE: min(8, len(players_by_position.get(Position.TE, []))),
Position.DEF: min(5, len(players_by_position.get(Position.DEF, []))),
}
combinations = []
# Generate combinations with FLEX consideration
qb_players = players_by_position.get(Position.QB, [])[
: max_players_per_position[Position.QB]
]
rb_players = players_by_position.get(Position.RB, [])[
: max_players_per_position[Position.RB]
]
wr_players = players_by_position.get(Position.WR, [])[
: max_players_per_position[Position.WR]
]
te_players = players_by_position.get(Position.TE, [])[
: max_players_per_position[Position.TE]
]
def_players = players_by_position.get(Position.DEF, [])[
: max_players_per_position[Position.DEF]
]
# Use concurrent processing for combination generation
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
# Generate core position combinations
core_combinations = list(
itertools.product(
itertools.combinations(qb_players, 1),
itertools.combinations(rb_players, 2),
itertools.combinations(wr_players, 3),
itertools.combinations(te_players, 1),
itertools.combinations(def_players, 1),
)
)
# Add FLEX players (can be RB, WR, or TE)
flex_players = rb_players + wr_players + te_players
for core_combo in core_combinations[:10000]: # Limit for performance
qbs, rbs, wrs, tes, defs = core_combo
used_players = set(list(qbs) + list(rbs) + list(wrs) + list(tes) + list(defs))
# Add available FLEX players
available_flex = [p for p in flex_players if p not in used_players]
for flex_player in available_flex[:3]: # Top 3 FLEX options
lineup_players = (
list(qbs) + list(rbs) + list(wrs) + list(tes) + list(defs) + [flex_player]
)
# Quick salary check for pruning
if self._quick_salary_check(lineup_players, constraints):
combinations.append(lineup_players)
return combinations
def _quick_salary_check(self, players: List[Player], constraints: LineupConstraints) -> bool:
"""Quick salary feasibility check for pruning."""
total_salary = 0
for player in players:
if player.value_metrics:
salary = (
player.value_metrics.draftkings_salary
or player.value_metrics.fanduel_salary
or player.value_metrics.yahoo_salary
or 5000
)
total_salary += salary
else:
total_salary += 5000
return total_salary <= constraints.salary_cap
async def _evaluate_lineup_combination(
self,
players: List[Player],
constraints: LineupConstraints,
weights: OptimizationWeights,
objective: OptimizationObjective,
) -> Tuple[Optional[Lineup], float]:
"""Evaluate a specific player combination as a lineup."""
try:
# Create lineup from players
lineup = self._create_lineup_from_players(players, constraints)
# Validate constraints
violations = lineup.validate_against_constraints(constraints)
if violations:
return None, float("-inf")
# Calculate multi-objective score
score = await self._calculate_lineup_score(lineup, weights, objective)
self.optimization_stats["total_evaluations"] += 1
return lineup, score
except Exception as e:
self.logger.debug(f"Error evaluating combination: {e}")
return None, float("-inf")
def _create_lineup_from_players(
self, players: List[Player], constraints: LineupConstraints
) -> Lineup:
"""Create a Lineup object from a list of players."""
# Map players to positions (simplified DraftKings structure)
position_map = [
Position.QB,
Position.RB,
Position.RB,
Position.WR,
Position.WR,
Position.WR,
Position.TE,
Position.RB,
Position.DEF,
]
slots = []
total_salary = 0
total_points = Decimal("0")
for i, player in enumerate(players):
position = position_map[i] if i < len(position_map) else player.position
salary = self._get_player_salary_for_optimization(player)
slot = LineupSlot(position=position, player=player, salary_used=salary)
slots.append(slot)
total_salary += salary
if player.projections:
total_points += player.projections.projected_fantasy_points
return Lineup(
lineup_type=LineupType.DRAFTKINGS,
slots=slots,
total_salary=total_salary,
salary_remaining=constraints.salary_cap - total_salary,
salary_cap=constraints.salary_cap,
total_projected_points=total_points,
confidence_score=Decimal("0.8"),
)
def _get_player_salary_for_optimization(self, player: Player) -> int:
"""Get player salary for optimization calculations."""
if player.value_metrics:
return (
player.value_metrics.draftkings_salary
or player.value_metrics.fanduel_salary
or player.value_metrics.yahoo_salary
or 5000
)
return 5000
async def _calculate_lineup_score(
self, lineup: Lineup, weights: OptimizationWeights, objective: OptimizationObjective
) -> float:
"""Calculate multi-objective score for a lineup."""
# Base projected points
points_score = float(lineup.total_projected_points)
# Value score (points per $1000)
value_score = float(lineup.get_salary_efficiency()) if lineup.total_salary > 0 else 0
# Ownership score (lower is better for contrarian plays)
ownership_score = 100 - float(lineup.projected_ownership or 50)
# Ceiling and floor scores
ceiling_score = float(lineup.ceiling_points or lineup.total_projected_points)
floor_score = float(lineup.floor_points or lineup.total_projected_points * Decimal("0.7"))
# Correlation bonus
correlation_score = await self._calculate_correlation_score(lineup)
# Variance penalty
variance_penalty = self._calculate_variance_penalty(lineup)
# Combine scores based on objective and weights
if objective == OptimizationObjective.MAXIMIZE_POINTS:
score = points_score
elif objective == OptimizationObjective.MAXIMIZE_VALUE:
score = value_score * 10 # Scale to similar range
elif objective == OptimizationObjective.MINIMIZE_OWNERSHIP:
score = ownership_score
elif objective == OptimizationObjective.MAXIMIZE_CEILING:
score = ceiling_score
elif objective == OptimizationObjective.MAXIMIZE_FLOOR:
score = floor_score
else: # BALANCED
score = (
points_score * weights.points
+ value_score * weights.value * 10
+ ownership_score * weights.ownership
+ ceiling_score * weights.ceiling
+ floor_score * weights.floor
+ correlation_score * weights.correlation
- variance_penalty * weights.variance_penalty
)
return score
async def _calculate_correlation_score(self, lineup: Lineup) -> float:
"""Calculate correlation bonus for stacked players."""
correlation_score = 0.0
players = lineup.get_players()
# QB-WR correlation bonus
qbs = lineup.get_players_by_position(Position.QB)
wrs = lineup.get_players_by_position(Position.WR)
for qb in qbs:
for wr in wrs:
if qb.team == wr.team:
correlation_score += 2.0 # Same team bonus
# RB-DEF negative correlation penalty
rbs = lineup.get_players_by_position(Position.RB)
defs = lineup.get_players_by_position(Position.DEF)
for rb in rbs:
for defense in defs:
if rb.team == defense.team:
correlation_score -= 1.0 # Same team penalty
return correlation_score
def _calculate_variance_penalty(self, lineup: Lineup) -> float:
"""Calculate variance penalty for high-risk lineups."""
variance_penalty = 0.0
# Penalty for multiple players from same team (except stacks)
team_exposure = lineup.get_team_exposure()
for team, count in team_exposure.items():
if count > 3: # More than 3 from same team
variance_penalty += (count - 3) * 0.5
return variance_penalty
async def _initialize_population(
self, players: List[Player], constraints: LineupConstraints, config: GeneticAlgorithmConfig
) -> List[LineupChromosome]:
"""Initialize genetic algorithm population."""
population = []
players_by_position = self._group_players_by_position(players)
# Create diverse initial population
for _ in range(config.population_size):
try:
chromosome_players = await self._create_random_valid_lineup(
players_by_position, constraints
)
chromosome = LineupChromosome(chromosome_players, constraints)
population.append(chromosome)
except Exception as e:
self.logger.debug(f"Failed to create random lineup: {e}")
continue
# Ensure minimum population size
while len(population) < config.population_size // 2:
try:
chromosome_players = await self._create_random_valid_lineup(
players_by_position, constraints
)
chromosome = LineupChromosome(chromosome_players, constraints)
population.append(chromosome)
except:
break
return population
async def _create_random_valid_lineup(
self, players_by_position: Dict[Position, List[Player]], constraints: LineupConstraints
) -> List[Player]:
"""Create a random valid lineup for genetic algorithm initialization."""
lineup_players = []
remaining_salary = constraints.salary_cap
# Standard DraftKings positions
positions_needed = [
Position.QB,
Position.RB,
Position.RB,
Position.WR,
Position.WR,
Position.WR,
Position.TE,
Position.RB,
Position.DEF,
]
for position in positions_needed:
available_players = players_by_position.get(position, [])
# Filter by remaining salary
affordable_players = [
p
for p in available_players
if p not in lineup_players
and self._get_player_salary_for_optimization(p) <= remaining_salary
]
if not affordable_players:
# Fallback to cheapest available
affordable_players = [p for p in available_players if p not in lineup_players]
if affordable_players:
affordable_players = [
min(
affordable_players,
key=lambda x: self._get_player_salary_for_optimization(x),
)
]
if affordable_players:
# Weighted random selection (higher projection = higher probability)
weights = [
float(p.projections.projected_fantasy_points) if p.projections else 1.0
for p in affordable_players
]
if sum(weights) > 0:
player = random.choices(affordable_players, weights=weights)[0]
else:
player = random.choice(affordable_players)
lineup_players.append(player)
remaining_salary -= self._get_player_salary_for_optimization(player)
return lineup_players
async def _evaluate_population_fitness(
self,
population: List[LineupChromosome],
weights: OptimizationWeights,
objective: OptimizationObjective,
) -> None:
"""Evaluate fitness for entire population in parallel."""
# Create evaluation tasks
tasks = []
for chromosome in population:
task = self._evaluate_chromosome_fitness(chromosome, weights, objective)
tasks.append(task)
# Execute in parallel batches
batch_size = min(100, len(tasks))
for i in range(0, len(tasks), batch_size):
batch = tasks[i : i + batch_size]
await asyncio.gather(*batch)
async def _evaluate_chromosome_fitness(
self,
chromosome: LineupChromosome,
weights: OptimizationWeights,
objective: OptimizationObjective,
) -> None:
"""Evaluate fitness for a single chromosome."""
try:
lineup = chromosome.to_lineup()
# Check constraint violations
violations = lineup.validate_against_constraints(chromosome.constraints)
if violations:
chromosome.fitness = float("-inf")
chromosome.violations = violations
return
# Calculate fitness score
fitness = await self._calculate_lineup_score(lineup, weights, objective)
chromosome.fitness = fitness
chromosome.violations = []
except Exception as e:
chromosome.fitness = float("-inf")
chromosome.violations = [f"Evaluation error: {str(e)}"]
async def _create_next_generation(
self,
population: List[LineupChromosome],
players: List[Player],
constraints: LineupConstraints,
config: GeneticAlgorithmConfig,
) -> List[LineupChromosome]:
"""Create next generation through selection, crossover, and mutation."""
# Sort population by fitness
population.sort(key=lambda x: x.fitness or float("-inf"), reverse=True)
next_generation = []
# Elitism: Keep top performers
elite_count = int(config.population_size * config.elitism_rate)
next_generation.extend(population[:elite_count])
# Generate offspring through crossover and mutation
while len(next_generation) < config.population_size:
# Tournament selection
parent1 = self._tournament_selection(population, config.tournament_size)
parent2 = self._tournament_selection(population, config.tournament_size)
# Crossover
if random.random() < config.crossover_rate:
offspring = await self._crossover(parent1, parent2, constraints)
else:
offspring = parent1 if parent1.fitness > parent2.fitness else parent2
# Mutation
if random.random() < config.mutation_rate:
offspring = await self._mutate(offspring, players, constraints)
next_generation.append(offspring)
return next_generation[: config.population_size]
def _tournament_selection(
self, population: List[LineupChromosome], tournament_size: int
) -> LineupChromosome:
"""Select parent using tournament selection."""
tournament = random.sample(population, min(tournament_size, len(population)))
return max(tournament, key=lambda x: x.fitness or float("-inf"))
async def _crossover(
self, parent1: LineupChromosome, parent2: LineupChromosome, constraints: LineupConstraints
) -> LineupChromosome:
"""Create offspring through crossover."""
try:
# Position-aware crossover
offspring_players = []
for i in range(len(parent1.players)):
if random.random() < 0.5:
offspring_players.append(parent1.players[i])
else:
offspring_players.append(parent2.players[i])
# Ensure no duplicate players
seen_players = set()
unique_players = []
for player in offspring_players:
if player.id not in seen_players:
unique_players.append(player)
seen_players.add(player.id)
# If we have duplicates, fill from parents
if len(unique_players) < len(offspring_players):
all_parent_players = parent1.players + parent2.players
for player in all_parent_players:
if len(unique_players) >= len(offspring_players):
break
if player.id not in seen_players:
unique_players.append(player)
seen_players.add(player.id)
return LineupChromosome(unique_players[: len(parent1.players)], constraints)
except Exception:
# Return better parent if crossover fails
return parent1 if parent1.fitness > parent2.fitness else parent2
async def _mutate(
self, chromosome: LineupChromosome, players: List[Player], constraints: LineupConstraints
) -> LineupChromosome:
"""Mutate chromosome by replacing random players."""
try:
mutated_players = chromosome.players.copy()
# Replace 1-2 random players
num_mutations = random.randint(1, 2)
positions_to_mutate = random.sample(range(len(mutated_players)), num_mutations)
for pos_idx in positions_to_mutate:
current_player = mutated_players[pos_idx]
position = current_player.position
# Find replacement candidates
position_players = [
p for p in players if p.position == position and p.id != current_player.id
]
if position_players:
# Filter by salary constraint
current_salary = sum(
self._get_player_salary_for_optimization(p) for p in mutated_players
)
current_player_salary = self._get_player_salary_for_optimization(current_player)
available_salary = (
constraints.salary_cap - current_salary + current_player_salary
)
affordable_players = [
p
for p in position_players
if self._get_player_salary_for_optimization(p) <= available_salary
]
if affordable_players:
new_player = random.choice(affordable_players)
mutated_players[pos_idx] = new_player
return LineupChromosome(mutated_players, constraints)
except Exception:
return chromosome
async def _create_alternative_lineup(
self,
players: List[Player],
constraints: LineupConstraints,
weights: OptimizationWeights,
objective: OptimizationObjective,
optimal_lineup: Lineup,
reason: str,
) -> Optional[LineupAlternative]:
"""Create an alternative lineup with different optimization focus."""
try:
# Modify constraints to force different players
modified_constraints = LineupConstraints(
salary_cap=constraints.salary_cap,
position_requirements=constraints.position_requirements,
max_players_per_team=constraints.max_players_per_team,
excluded_players=(constraints.excluded_players or [])
+ [p.id for p in optimal_lineup.get_players()[:3]],
)
alternative_lineup = await self._quick_optimization(
players, modified_constraints, objective
)
if alternative_lineup:
# Calculate differences
point_diff = (
alternative_lineup.total_projected_points
- optimal_lineup.total_projected_points
)
salary_diff = alternative_lineup.total_salary - optimal_lineup.total_salary
ownership_diff = (alternative_lineup.projected_ownership or Decimal("50")) - (
optimal_lineup.projected_ownership or Decimal("50")
)
return LineupAlternative(
lineup=alternative_lineup,
reason=reason,
point_difference=point_diff,
salary_difference=salary_diff,
ownership_difference=ownership_diff,
confidence=Decimal("0.7"),
)
except Exception as e:
self.logger.debug(f"Failed to create alternative: {e}")
return None
async def _quick_optimization(
self,
players: List[Player],
constraints: LineupConstraints,
objective: OptimizationObjective = OptimizationObjective.MAXIMIZE_POINTS,
) -> Optional[Lineup]:
"""Quick optimization for alternatives and evaluations."""
try:
# Use simplified greedy optimization for speed
return await self._greedy_optimization(players, constraints, objective)
except Exception:
return None
async def _greedy_optimization(
self,
players: List[Player],
constraints: LineupConstraints,
objective: OptimizationObjective,
) -> Optional[Lineup]:
"""Fast greedy optimization approach."""
# Sort players by optimization criteria
if objective == OptimizationObjective.MAXIMIZE_POINTS:
players.sort(
key=lambda p: (
p.projections.projected_fantasy_points if p.projections else Decimal("0")
),
reverse=True,
)
elif objective == OptimizationObjective.MAXIMIZE_VALUE:
players.sort(key=lambda p: p.get_projected_value() or Decimal("0"), reverse=True)
elif objective == OptimizationObjective.MINIMIZE_OWNERSHIP:
players.sort(
key=lambda p: (
p.value_metrics.ownership_percentage if p.value_metrics else Decimal("50")
)
)
# Greedy selection by position
selected_players = []
remaining_salary = constraints.salary_cap
positions_needed = [
Position.QB,
Position.RB,
Position.RB,
Position.WR,
Position.WR,
Position.WR,
Position.TE,
Position.RB,
Position.DEF,
]
for position in positions_needed:
best_player = None
for player in players:
if (
player.position == position
and player not in selected_players
and self._get_player_salary_for_optimization(player) <= remaining_salary
):
if constraints.excluded_players and player.id in constraints.excluded_players:
continue
best_player = player
break
if best_player:
selected_players.append(best_player)
remaining_salary -= self._get_player_salary_for_optimization(best_player)
else:
return None # Cannot complete lineup
if len(selected_players) == 9:
return self._create_lineup_from_players(selected_players, constraints)
return None
async def _evaluate_replacement_player(
self, injured_player: Player, replacement: Player, constraints: LineupConstraints
) -> Tuple[float, str]:
"""Evaluate how well a replacement player fills the injured player's role."""
# Basic point comparison
injured_points = (
injured_player.projections.projected_fantasy_points
if injured_player.projections
else Decimal("0")
)
replacement_points = (
replacement.projections.projected_fantasy_points
if replacement.projections
else Decimal("0")
)
point_difference = float(replacement_points - injured_points)
# Factor in matchup quality
matchup_factor = 1.0
if replacement.projections and replacement.projections.matchup_rating:
if "favorable" in replacement.projections.matchup_rating.lower():
matchup_factor = 1.2
elif "difficult" in replacement.projections.matchup_rating.lower():
matchup_factor = 0.8
# Factor in consistency
consistency_factor = self._calculate_consistency_factor(replacement)
score = point_difference * matchup_factor * consistency_factor
reason = f"{point_difference:+.1f} pts vs injured player, {matchup_factor:.1f}x matchup, {consistency_factor:.1f}x consistency"
return score, reason
def _calculate_upside_factor(self, player: Player, weeks_ahead: int) -> float:
"""Calculate upside factor based on player's ceiling potential."""
if not player.projections:
return 1.0
base_projection = float(player.projections.projected_fantasy_points)
ceiling_projection = float(
player.projections.ceiling_points or player.projections.projected_fantasy_points
)
if base_projection > 0:
upside_ratio = ceiling_projection / base_projection
return min(upside_ratio, 2.0) # Cap at 2x
return 1.0
def _calculate_consistency_factor(self, player: Player) -> float:
"""Calculate consistency factor based on floor vs projection."""
if not player.projections:
return 1.0
base_projection = float(player.projections.projected_fantasy_points)
floor_projection = float(
player.projections.floor_points
or player.projections.projected_fantasy_points * Decimal("0.7")
)
if base_projection > 0:
floor_ratio = floor_projection / base_projection
return max(floor_ratio, 0.5) # Minimum 0.5x
return 1.0
def _get_strategy_weights(self, strategy: OptimizationStrategy) -> OptimizationWeights:
"""Get optimization weights based on strategy."""
if strategy == OptimizationStrategy.MAX_POINTS:
return OptimizationWeights(points=1.0, value=0.0, ownership=0.0)
elif strategy == OptimizationStrategy.MAX_VALUE:
return OptimizationWeights(points=0.3, value=0.7, ownership=0.0)
elif strategy == OptimizationStrategy.LOW_OWNERSHIP:
return OptimizationWeights(points=0.4, value=0.2, ownership=0.4)
elif strategy == OptimizationStrategy.CONTRARIAN:
return OptimizationWeights(points=0.3, value=0.2, ownership=0.5)
elif strategy == OptimizationStrategy.SAFE:
return OptimizationWeights(points=0.5, value=0.2, ownership=0.0, floor=0.3)
elif strategy == OptimizationStrategy.GPP:
return OptimizationWeights(points=0.4, value=0.2, ownership=0.2, ceiling=0.2)
else: # BALANCED
return OptimizationWeights(points=0.5, value=0.3, ownership=0.2)
def _estimate_search_space(self, players: List[Player]) -> int:
"""Estimate the size of the optimization search space."""
players_by_position = self._group_players_by_position(players)
# Estimate combinations for standard DraftKings lineup
qb_count = len(players_by_position.get(Position.QB, []))
rb_count = len(players_by_position.get(Position.RB, []))
wr_count = len(players_by_position.get(Position.WR, []))
te_count = len(players_by_position.get(Position.TE, []))
def_count = len(players_by_position.get(Position.DEF, []))
# Rough estimate: QB(1) * RB(2) * WR(3) * TE(1) * DEF(1) * FLEX options
if all([qb_count, rb_count >= 2, wr_count >= 3, te_count, def_count]):
from math import comb
estimate = (
qb_count
* comb(rb_count, 2)
* comb(wr_count, 3)
* te_count
* def_count
* max(1, rb_count + wr_count + te_count - 6) # FLEX options
)
return estimate
return 0
def _create_recommendation(
self,
optimal_lineup: Lineup,
alternatives: List[LineupAlternative],
strategy: OptimizationStrategy,
weights: OptimizationWeights,
) -> LineupRecommendation:
"""Create comprehensive lineup recommendation."""
# Generate reasoning
reasoning = self._generate_reasoning(optimal_lineup, strategy, weights)
# Key factors
key_factors = [
f"Projected {optimal_lineup.total_projected_points} points",
f"${optimal_lineup.salary_remaining} salary remaining",
f"{optimal_lineup.projected_ownership or 50:.1f}% projected ownership",
]
# Risk assessment
risk_level = self._assess_risk_level(optimal_lineup)
upside_potential = self._assess_upside_potential(optimal_lineup)
floor_assessment = self._assess_floor_potential(optimal_lineup)
# Contest recommendations
recommended_contests = self._get_contest_recommendations(optimal_lineup, strategy)
return LineupRecommendation(
optimal_lineup=optimal_lineup,
alternatives=alternatives,
reasoning=reasoning,
key_factors=key_factors,
strategy=strategy,
contest_type=recommended_contests[0] if recommended_contests else "GPP",
risk_level=risk_level,
upside_potential=upside_potential,
floor_assessment=floor_assessment,
recommended_contest_types=recommended_contests,
week=1, # Would be dynamic
season=2024, # Would be dynamic
overall_confidence=Decimal("0.8"),
)
def _generate_reasoning(
self, lineup: Lineup, strategy: OptimizationStrategy, weights: OptimizationWeights
) -> str:
"""Generate reasoning text for the lineup recommendation."""
reasoning_parts = []
# Strategy-specific reasoning
if strategy == OptimizationStrategy.MAX_POINTS:
reasoning_parts.append("Optimized for maximum projected points")
elif strategy == OptimizationStrategy.MAX_VALUE:
reasoning_parts.append("Optimized for salary value efficiency")
elif strategy == OptimizationStrategy.LOW_OWNERSHIP:
reasoning_parts.append("Designed for contrarian, low-ownership plays")
# Team stacking
team_exposure = lineup.get_team_exposure()
high_exposure_teams = [team for team, count in team_exposure.items() if count >= 2]
if high_exposure_teams:
reasoning_parts.append(f"Features team stacks from: {', '.join(high_exposure_teams)}")
# Salary efficiency
efficiency = lineup.get_salary_efficiency()
reasoning_parts.append(f"Salary efficiency: {efficiency:.2f} points per $1K")
return ". ".join(reasoning_parts) + "."
def _assess_risk_level(self, lineup: Lineup) -> str:
"""Assess overall risk level of the lineup."""
# Factors: ownership, variance, team concentration
ownership = float(lineup.projected_ownership or 50)
if ownership > 70:
return "High" # High-owned = higher risk of not differentiating
elif ownership < 30:
return "High" # Very low-owned = higher bust risk
else:
return "Medium"
def _assess_upside_potential(self, lineup: Lineup) -> str:
"""Assess upside potential of the lineup."""
ceiling = lineup.ceiling_points or lineup.total_projected_points * Decimal("1.3")
projection = lineup.total_projected_points
upside_ratio = float(ceiling / projection) if projection > 0 else 1.0
if upside_ratio > 1.4:
return "Very High"
elif upside_ratio > 1.2:
return "High"
else:
return "Medium"
def _assess_floor_potential(self, lineup: Lineup) -> str:
"""Assess floor/safety of the lineup."""
floor = lineup.floor_points or lineup.total_projected_points * Decimal("0.7")
projection = lineup.total_projected_points
floor_ratio = float(floor / projection) if projection > 0 else 0.7
if floor_ratio > 0.8:
return "Very Safe"
elif floor_ratio > 0.7:
return "Safe"
else:
return "Volatile"
def _get_contest_recommendations(
self, lineup: Lineup, strategy: OptimizationStrategy
) -> List[str]:
"""Get recommended contest types for the lineup."""
contests = []
if strategy in [OptimizationStrategy.SAFE, OptimizationStrategy.CASH_GAME]:
contests.extend(["Cash Games", "Double-ups", "50/50s"])
if strategy in [OptimizationStrategy.GPP, OptimizationStrategy.CONTRARIAN]:
contests.extend(["GPPs", "Tournaments", "Large-field contests"])
if strategy == OptimizationStrategy.BALANCED:
contests.extend(["Cash Games", "Small-field GPPs"])
return contests or ["GPPs"]
# CPU-bound optimization helper functions for ProcessPoolExecutor
@staticmethod
def _evaluate_combination_batch(
combinations_batch: List[List[Player]],
constraints_dict: Dict[str, Any],
weights_dict: Dict[str, float],
objective: str,
) -> List[Tuple[Optional[Dict], float]]:
"""Evaluate a batch of combinations in a separate process."""
results = []
for combination in combinations_batch:
try:
# Simplified evaluation for CPU-bound processing
total_salary = sum(
player.value_metrics.draftkings_salary or 5000
for player in combination
if player.value_metrics
)
total_points = sum(
float(player.projections.projected_fantasy_points)
for player in combination
if player.projections
)
if total_salary <= constraints_dict["salary_cap"]:
score = total_points # Simplified scoring
lineup_dict = {
"players": [{"id": p.id, "name": p.name} for p in combination],
"total_salary": total_salary,
"total_points": total_points,
}
results.append((lineup_dict, score))
else:
results.append((None, float("-inf")))
except Exception:
results.append((None, float("-inf")))
return results
```
--------------------------------------------------------------------------------
/fastmcp_server.py:
--------------------------------------------------------------------------------
```python
from __future__ import annotations
"""FastMCP-compatible fantasy football server entry point.
This module wraps the existing Yahoo Fantasy Football tooling defined in
``fantasy_football_multi_league`` and exposes it through the FastMCP
``@server.tool`` decorator so it can be deployed on fastmcp.cloud.
"""
import json
import os
from collections.abc import Iterable
from dataclasses import asdict, is_dataclass
from typing import Any, Awaitable, Callable, Dict, Literal, Optional, Sequence, Union
from fastmcp import Context, FastMCP
from mcp.types import ContentBlock, TextContent
import fantasy_football_multi_league
# REMOVED: enhanced_mcp_tools imports - no longer using wrapper tools
# Remove explicit typing to avoid type conflicts with evolving MCP types
_legacy_call_tool = fantasy_football_multi_league.call_tool
_legacy_refresh_token = fantasy_football_multi_league.refresh_yahoo_token
server = FastMCP(
name="fantasy-football",
instructions=(
"Yahoo Fantasy Football operations including league discovery, roster "
"analysis, waiver insights, draft tools, and Reddit sentiment checks. "
"Set the YAHOO_* environment variables before starting the server."
),
)
_TOOL_PROMPTS: Dict[str, str] = {
"ff_get_leagues": (
"🏈 LEAGUE DISCOVERY - List all Yahoo fantasy leagues for the user. "
"Takes NO parameters. Use FIRST to get league_key values. "
"For player searches use ff_get_players or ff_get_waiver_wire."
),
"ff_get_league_info": (
"📋 Get league configuration and settings. "
"Parameters: league_key only. Returns scoring type and your team summary."
),
"ff_get_standings": (
"🏆 Get current league standings. "
"Parameters: league_key only. Returns ranks, records, points for all teams."
),
"ff_get_roster": (
"Get roster data with configurable detail levels. Use data_level='basic' for "
"quick roster info, 'standard' for roster + projections, or 'full' for "
"comprehensive analysis with external data sources and enhanced insights."
),
"ff_get_matchup": (
"🆚 Get weekly matchup for your team. "
"Parameters: league_key (required), week (optional). Returns opponent and projections."
),
"ff_get_players": (
"Research free agents or player pools for waiver pickups by filtering "
"Yahoo players by position and limiting the result count. Accepts optional "
"parameters for enhanced analysis similar to roster data."
),
"ff_compare_teams": (
"Contrast two league rosters side-by-side to evaluate trades or matchup "
"advantages. Provide both Yahoo team keys."
),
"ff_build_lineup": (
"Build optimal lineup from your roster using strategy-based optimization and positional constraints."
),
"ff_refresh_token": (
"🔑 Refresh Yahoo OAuth token. " "NO parameters. Use when API returns 401 errors."
),
"ff_get_api_status": (
"📊 Check API health and rate limits. "
"NO parameters. Returns cache metrics and throttling status."
),
"ff_clear_cache": (
"Clear cached Yahoo responses to force the next call to fetch fresh "
"data. Optionally specify a pattern to target certain entries."
),
"ff_get_draft_results": (
"Retrieve the draft board and pick summaries for every team in a league "
"after the draft has completed."
),
"ff_get_waiver_wire": (
"List waiver-wire candidates sorted by rank, points, or trends to aid "
"mid-season roster moves."
),
"ff_get_draft_rankings": (
"Access Yahoo pre-draft rankings and ADP information for planning "
"upcoming drafts, filtered by position if desired."
),
"ff_get_draft_recommendation": (
"Recommend players to draft at the current or upcoming pick based on "
"your strategy and league context."
),
"ff_analyze_draft_state": (
"Evaluate the evolving draft board for your team to highlight "
"positional needs and strategy adjustments."
),
"ff_analyze_reddit_sentiment": (
"Summarize recent Reddit sentiment and engagement around one or more "
"players to complement scouting insights."
),
}
def _tool_meta(name: str) -> Dict[str, str]:
"""Helper to attach consistent prompt metadata to each tool."""
return {"prompt": _TOOL_PROMPTS[name]}
async def _call_legacy_tool(
name: str,
*,
ctx: Context | None = None,
**arguments: Any,
) -> Dict[str, Any]:
"""Delegate to the legacy MCP tool implementation and parse its JSON payload."""
filtered_args = {key: value for key, value in arguments.items() if value is not None}
if ctx is not None:
await ctx.info(f"Calling legacy Yahoo tool: {name}")
raw_blocks = await _legacy_call_tool(name=name, arguments=filtered_args)
if raw_blocks is None:
blocks: Sequence[Any] = []
elif isinstance(raw_blocks, Iterable) and not isinstance(raw_blocks, (str, bytes, TextContent)):
blocks = list(raw_blocks)
else:
blocks = [raw_blocks]
if not blocks:
return {
"status": "error",
"message": "Legacy tool returned no response",
"tool": name,
"arguments": filtered_args,
}
def _coerce_text(block: Any) -> TextContent:
if isinstance(block, TextContent):
return block
if hasattr(block, "text") and isinstance(getattr(block, "text"), str):
return TextContent(type="text", text=getattr(block, "text"))
if is_dataclass(block) and not isinstance(block, type):
return TextContent(type="text", text=json.dumps(asdict(block)))
if hasattr(block, "data"):
data = getattr(block, "data")
if isinstance(data, bytes):
try:
data = data.decode("utf-8")
except Exception:
data = repr(data)
if isinstance(data, str):
return TextContent(type="text", text=data)
try:
return TextContent(type="text", text=json.dumps(block, default=str))
except Exception:
return TextContent(type="text", text=str(block))
responses = [_coerce_text(block) for block in blocks]
first = responses[0]
payload = getattr(first, "text", "")
# Instrumentation: detect raw '0' / suspiciously tiny payloads that break higher layers
if payload.strip() == "0":
diag = {
"status": "error",
"message": "Legacy tool returned sentinel '0' string instead of JSON",
"tool": name,
"arguments": filtered_args,
"raw": payload,
"stage": "_call_legacy_tool:raw_payload_zero",
}
if ctx is not None:
await ctx.info(f"[diagnostic] Detected raw '0' payload from legacy tool: {name}")
return diag
if not payload:
return {
"status": "error",
"message": "Legacy tool returned an empty payload",
"tool": name,
"arguments": filtered_args,
}
try:
return json.loads(payload)
except json.JSONDecodeError:
return {
"status": "error",
"message": "Could not parse legacy response as JSON",
"tool": name,
"arguments": filtered_args,
"raw": payload,
}
@server.tool(
name="ff_get_leagues",
description=(
"🏈 LEAGUE DISCOVERY - Get list of your Yahoo fantasy leagues. "
"NO parameters required (no position/count/sort). "
"Use this FIRST to get league_key values. "
"For player searches use ff_get_players or ff_get_waiver_wire."
),
meta=_tool_meta("ff_get_leagues"),
)
async def ff_get_leagues(ctx: Context) -> Dict[str, Any]:
"""
Discover all Yahoo fantasy football leagues for the authenticated user.
⚠️ IMPORTANT: This tool takes NO search parameters!
- NO position, count, sort, week, or team_key parameters
- This is for LEAGUE DISCOVERY only, not player searches
For player searches use:
- ff_get_players → Search available players by position
- ff_get_waiver_wire → Waiver wire analysis with rankings
- ff_get_roster → Get YOUR team's current roster
Returns:
Dict with total_leagues count and list of league summaries
"""
return await _call_legacy_tool("ff_get_leagues", ctx=ctx)
@server.tool(
name="ff_get_league_info",
description=(
"📋 Get league configuration and settings. "
"Parameters: league_key (required only). "
"Returns scoring type, roster requirements, and your team summary."
),
meta=_tool_meta("ff_get_league_info"),
)
async def ff_get_league_info(
ctx: Context,
league_key: str,
) -> Dict[str, Any]:
"""
Retrieve metadata about a single Yahoo league.
Args:
league_key: League identifier (required)
Returns:
Dict with league settings, scoring type, and your team info
"""
return await _call_legacy_tool(
"ff_get_league_info",
ctx=ctx,
league_key=league_key,
)
@server.tool(
name="ff_get_roster",
description=(
"⚠️ Get YOUR TEAM'S current roster (YOUR players only). "
"DO NOT use this to search for available players! "
"Parameters: league_key, team_key, week, data_level, include_projections, include_external_data, include_analysis. "
"For available players use ff_get_players or ff_get_waiver_wire."
),
meta=_tool_meta("ff_get_roster"),
)
async def ff_get_roster(
ctx: Context,
league_key: str,
team_key: Optional[str] = None,
week: Optional[int] = None,
include_projections: bool = True,
include_external_data: bool = True,
include_analysis: bool = True,
data_level: Optional[Literal["basic", "standard", "full"]] = None,
) -> Dict[str, Any]:
"""
Get YOUR TEAM'S roster with configurable detail levels.
⚠️ IMPORTANT: This tool ONLY gets YOUR roster, not available players.
- To search available players by position → use ff_get_players
- For waiver wire pickups with rankings → use ff_get_waiver_wire
This tool does NOT accept: position, count, sort, include_expert_analysis
Args:
league_key: League identifier
team_key: Team identifier (optional, defaults to authenticated user's team)
week: Week number for projections (optional, defaults to current week)
include_projections: Include Yahoo and/or Sleeper projections
include_external_data: Include Sleeper rankings, matchup analysis, trending data
include_analysis: Include enhanced player analysis and recommendations
data_level: "basic" (roster only), "standard" (+ projections), "full" (everything)
"""
# Ensure we have a valid data_level
if data_level is None:
data_level = "full"
# Determine effective settings based on data_level and explicit parameters
if data_level == "basic":
effective_projections = False
effective_external = False
effective_analysis = False
elif data_level == "standard":
effective_projections = True
effective_external = False
effective_analysis = False
else: # "full"
effective_projections = True
effective_external = True
effective_analysis = True
# Explicit parameters override data_level defaults
if not include_projections:
effective_projections = False
if not include_external_data:
effective_external = False
if not include_analysis:
effective_analysis = False
# Informational logging for the selected mode
if ctx:
if not any([effective_projections, effective_external, effective_analysis]):
await ctx.info("Using basic roster data (legacy mode)")
else:
await ctx.info(
"Using enhanced roster data "
f"(projections: {effective_projections}, external: {effective_external}, analysis: {effective_analysis})"
)
try:
result = await _call_legacy_tool(
"ff_get_roster",
ctx=ctx,
league_key=league_key,
team_key=team_key,
week=week,
include_projections=effective_projections,
include_external_data=effective_external,
include_analysis=effective_analysis,
data_level=data_level,
)
return result
except Exception as exc:
return {
"status": "error",
"message": f"Enhanced roster fetch failed: {exc}",
"fallback_suggestion": "Try using data_level='basic' for simple roster data",
}
@server.tool(
name="ff_get_standings",
description=(
"🏆 Get current league standings and team records. "
"Parameters: league_key (required only). "
"Returns rank, wins, losses, points for/against for all teams."
),
meta=_tool_meta("ff_get_standings"),
)
async def ff_get_standings(
ctx: Context,
league_key: str,
) -> Dict[str, Any]:
"""
Return the current standings table for a Yahoo league.
Args:
league_key: League identifier (required)
Returns:
Dict with sorted standings showing ranks, records, and points
"""
return await _call_legacy_tool("ff_get_standings", ctx=ctx, league_key=league_key)
@server.tool(
name="ff_get_matchup",
description=(
"🆚 Get weekly matchup for your team. "
"Parameters: league_key (required), week (optional, defaults to current). "
"Returns opponent info and projected scores."
),
meta=_tool_meta("ff_get_matchup"),
)
async def ff_get_matchup(
ctx: Context,
league_key: str,
week: Optional[int] = None,
) -> Dict[str, Any]:
"""
Retrieve matchup information for the authenticated team.
Args:
league_key: League identifier (required)
week: Week number (optional, defaults to current week)
Returns:
Dict with matchup data including opponent and projections
"""
return await _call_legacy_tool(
"ff_get_matchup",
ctx=ctx,
league_key=league_key,
week=week,
)
@server.tool(
name="ff_get_players",
description=(
"🔍 Search AVAILABLE players by position with count limit. "
"Use this to find free agents by position (QB, RB, WR, TE). "
"Parameters: league_key, position, count, week. "
"For YOUR roster use ff_get_roster. For waiver analysis use ff_get_waiver_wire."
),
meta=_tool_meta("ff_get_players"),
)
async def ff_get_players(
ctx: Context,
league_key: str,
position: Optional[str] = None,
count: int = 10,
week: Optional[int] = None,
team_key: Optional[str] = None,
data_level: Optional[Literal["basic", "standard", "full"]] = None,
include_analysis: Optional[bool] = None,
include_projections: Optional[bool] = None,
include_external_data: Optional[bool] = None,
) -> Dict[str, Any]:
"""
Enhanced player search with expert analysis and Sleeper integration.
Args:
league_key: League identifier
position: Filter by position (QB, RB, WR, TE, etc.)
count: Number of players to return
week: Week for analysis context
data_level: "basic" (names only), "standard" (+ stats), "full" (+ expert analysis)
include_analysis: Include expert tiers and recommendations
include_projections: Include projection data
include_external_data: Include Sleeper rankings and trending data
"""
# Default to enhanced mode for better player analysis
if data_level is None:
data_level = "full"
if include_analysis is None:
include_analysis = True
if include_external_data is None:
include_external_data = True
return await _call_legacy_tool(
"ff_get_players",
ctx=ctx,
league_key=league_key,
position=position,
count=count,
week=week,
team_key=team_key,
data_level=data_level,
include_analysis=include_analysis,
include_projections=include_projections,
include_external_data=include_external_data,
)
@server.tool(
name="ff_compare_teams",
description=(
"Compare the rosters of two teams in the same league to support trade "
"or matchup analysis. Provide both team keys."
),
meta=_tool_meta("ff_compare_teams"),
)
async def ff_compare_teams(
ctx: Context,
league_key: str,
team_key_a: str,
team_key_b: str,
) -> Dict[str, Any]:
return await _call_legacy_tool(
"ff_compare_teams",
ctx=ctx,
league_key=league_key,
team_key_a=team_key_a,
team_key_b=team_key_b,
)
@server.tool(
name="ff_build_lineup",
description=(
"Build optimal lineup from your roster using strategy-based optimization and positional constraints. "
"Uses advanced analytics including matchup analysis, player projections, and situational factors."
),
meta=_tool_meta("ff_build_lineup"),
)
async def ff_build_lineup(
ctx: Context,
league_key: str,
week: Optional[int] = None,
strategy: Literal["conservative", "aggressive", "balanced"] = "balanced",
debug: bool = False,
) -> Dict[str, Any]:
return await _call_legacy_tool(
"ff_build_lineup",
ctx=ctx,
league_key=league_key,
week=week,
strategy=strategy,
debug=debug,
)
@server.tool(
name="ff_refresh_token",
description=(
"🔑 Refresh Yahoo OAuth token. "
"NO parameters required. "
"Use when API calls return 401 authentication errors."
),
meta=_tool_meta("ff_refresh_token"),
)
async def ff_refresh_token(ctx: Context) -> Dict[str, Any]:
"""
Refresh the Yahoo OAuth access token.
⚠️ Takes NO parameters - automatic token refresh only
Returns:
Dict with token refresh status
"""
if ctx is not None:
await ctx.info("Refreshing Yahoo OAuth token")
return await _legacy_refresh_token()
@server.tool(
name="ff_get_api_status",
description=(
"📊 Check API health and rate limits. "
"NO parameters required. "
"Returns cache metrics and API throttling status."
),
meta=_tool_meta("ff_get_api_status"),
)
async def ff_get_api_status(ctx: Context) -> Dict[str, Any]:
"""
Inspect rate limiter and cache metrics for troubleshooting.
⚠️ Takes NO parameters - system diagnostic tool only
Returns:
Dict with API status, rate limits, and cache metrics
"""
return await _call_legacy_tool("ff_get_api_status", ctx=ctx)
@server.tool(
name="ff_clear_cache",
description=(
"Invalidate the Yahoo response cache. Optionally provide a pattern to "
"clear a subset of cached endpoints."
),
meta=_tool_meta("ff_clear_cache"),
)
async def ff_clear_cache(
ctx: Context,
pattern: Optional[str] = None,
) -> Dict[str, Any]:
return await _call_legacy_tool("ff_clear_cache", ctx=ctx, pattern=pattern)
@server.tool(
name="ff_get_draft_results",
description=(
"Fetch draft grades and pick positions for every team in a league to "
"review draft performance."
),
meta=_tool_meta("ff_get_draft_results"),
)
async def ff_get_draft_results(ctx: Context, league_key: str) -> Dict[str, Any]:
return await _call_legacy_tool("ff_get_draft_results", ctx=ctx, league_key=league_key)
@server.tool(
name="ff_get_waiver_wire",
description=(
"📊 Get waiver wire pickups with RANKINGS, SORTING, and expert analysis. "
"Use this for waiver priority decisions with sort options (rank/points/owned/trending). "
"Parameters: league_key, position, sort, count, include_expert_analysis. "
"For YOUR roster use ff_get_roster. For simple player search use ff_get_players."
),
meta=_tool_meta("ff_get_waiver_wire"),
)
async def ff_get_waiver_wire(
ctx: Context,
league_key: str,
position: Optional[str] = None,
sort: Literal["rank", "points", "owned", "trending"] = "rank",
count: int = 30,
week: Optional[int] = None,
team_key: Optional[str] = None,
include_expert_analysis: bool = True,
data_level: Optional[Literal["basic", "standard", "full"]] = None,
) -> Dict[str, Any]:
"""
Enhanced waiver wire analysis with expert recommendations.
Args:
league_key: League identifier
position: Filter by position (QB, RB, WR, TE, etc.) - defaults to "all"
sort: Sort method - "rank" (expert), "points" (season), "owned" (popularity), "trending" (hot pickups)
count: Number of players to return
week: Week for projections (optional, defaults to current)
team_key: Team key for context (optional)
include_expert_analysis: Include tiers, recommendations, and confidence scores
data_level: Data detail level ("basic", "standard", "full")
"""
# Default to enhanced mode for better waiver analysis, but basic mode if expert analysis disabled
if data_level is None:
data_level = "full" if include_expert_analysis else "basic"
# Handle position default - convert None to "all"
if position is None:
position = "all"
try:
# Map data_level to legacy parameters for backward compatibility
if data_level == "basic":
include_projections = False
include_external_data = False
include_analysis = False
elif data_level == "standard":
include_projections = True
include_external_data = False
include_analysis = False
else: # "full"
include_projections = True
include_external_data = True
include_analysis = include_expert_analysis
result = await _call_legacy_tool(
"ff_get_waiver_wire",
ctx=ctx,
league_key=league_key,
position=position,
sort=sort,
count=count,
week=week,
team_key=team_key,
include_projections=include_projections,
include_external_data=include_external_data,
include_analysis=include_analysis,
)
# Check if main server provided enhanced players
if include_expert_analysis and result.get("enhanced_players"):
# Main server handled the enhancement - use enhanced data
if ctx:
await ctx.info("Using enhanced waiver wire data from main server...")
# Replace basic players with enhanced players for better data
result["players"] = result["enhanced_players"]
# Ensure proper sorting based on request
if sort == "rank" and result["players"]:
# Sort by waiver_priority if available, else expert_confidence
if "waiver_priority" in result["players"][0]:
result["players"].sort(key=lambda x: x.get("waiver_priority", 0), reverse=True)
else:
result["players"].sort(
key=lambda x: x.get("expert_confidence", 0), reverse=True
)
elif sort == "trending":
result["players"].sort(key=lambda x: x.get("trending_score", 50), reverse=True)
elif include_expert_analysis and ctx:
await ctx.info("Expert analysis requested but not available from main server")
return result
except Exception as exc:
return {
"status": "error",
"message": f"Waiver wire analysis failed: {exc}",
"league_key": league_key,
}
@server.tool(
name="ff_get_draft_rankings",
description=(
"Access pre-draft Yahoo rankings and ADP data. Useful before or during "
"drafts to evaluate player tiers."
),
meta=_tool_meta("ff_get_draft_rankings"),
)
async def ff_get_draft_rankings(
ctx: Context,
league_key: Optional[str] = None,
position: Optional[str] = "all",
count: int = 50,
) -> Dict[str, Any]:
return await _call_legacy_tool(
"ff_get_draft_rankings",
ctx=ctx,
league_key=league_key,
position=position,
count=count,
)
@server.tool(
name="ff_get_draft_recommendation",
description=(
"Provide draft pick recommendations tailored to a strategy such as "
"balanced, aggressive, or conservative."
),
meta=_tool_meta("ff_get_draft_recommendation"),
)
async def ff_get_draft_recommendation(
ctx: Context,
league_key: str,
strategy: Literal["conservative", "aggressive", "balanced"] = "balanced",
num_recommendations: int = 10,
current_pick: Optional[int] = None,
) -> Dict[str, Any]:
return await _call_legacy_tool(
"ff_get_draft_recommendation",
ctx=ctx,
league_key=league_key,
strategy=strategy,
num_recommendations=num_recommendations,
current_pick=current_pick,
)
@server.tool(
name="ff_analyze_draft_state",
description=(
"Summarize the current draft landscape for your team, highlighting "
"positional needs and strategic advice."
),
meta=_tool_meta("ff_analyze_draft_state"),
)
async def ff_analyze_draft_state(
ctx: Context,
league_key: str,
strategy: Literal["conservative", "aggressive", "balanced"] = "balanced",
) -> Dict[str, Any]:
return await _call_legacy_tool(
"ff_analyze_draft_state",
ctx=ctx,
league_key=league_key,
strategy=strategy,
)
@server.tool(
name="ff_analyze_reddit_sentiment",
description=(
"Analyze recent Reddit chatter for one or more players to gauge public "
"sentiment, injury mentions, and engagement levels."
),
meta=_tool_meta("ff_analyze_reddit_sentiment"),
)
async def ff_analyze_reddit_sentiment(
ctx: Context,
players: Sequence[str],
time_window_hours: int = 48,
) -> Dict[str, Any]:
return await _call_legacy_tool(
"ff_analyze_reddit_sentiment",
ctx=ctx,
players=list(players),
time_window_hours=time_window_hours,
)
# ============================================================================
# ENHANCED TOOLS - Advanced decision-making capabilities for client LLMs
# ============================================================================
# REMOVED: ff_get_roster_with_projections_wrapper - replaced by ff_get_roster with data_level='full'
# REMOVED: ff_analyze_lineup_options_wrapper - complex functionality can be achieved through ff_build_lineup
# REMOVED: ff_compare_players_wrapper - player comparison can be done through ff_get_players and ff_get_waiver_wire
# REMOVED: ff_what_if_analysis_wrapper - scenario analysis can be done using ff_build_lineup with different strategies
# REMOVED: ff_get_decision_context_wrapper - context can be gathered through ff_get_league_info, ff_get_matchup, ff_get_standings
# ============================================================================
# PROMPTS - Reusable message templates for better LLM interactions
# ============================================================================
@server.prompt
def analyze_roster_strengths(league_key: str, team_key: str) -> str:
"""Generate a prompt for analyzing roster strengths and weaknesses."""
return f"""Please analyze the fantasy football roster for team {team_key} in league {league_key}.
Focus on:
1. Positional depth and strength
2. Starting lineup quality vs bench depth
3. Injury concerns and bye week coverage
4. Trade opportunities and waiver wire needs
5. Overall team competitiveness
Provide specific recommendations for improvement."""
@server.prompt
def draft_strategy_advice(strategy: str, league_size: int, pick_position: int) -> str:
"""Generate a prompt for draft strategy recommendations."""
return f"""Provide fantasy football draft strategy advice for:
- Strategy: {strategy}
- League size: {league_size} teams
- Draft position: {pick_position}
Include:
1. First 3 rounds strategy
2. Position priority order
3. Sleepers and value picks
4. Players to avoid
5. Late-round targets
6. PPR-specific considerations (pass-catching RBs, high-volume WRs)
Tailor the advice to the {strategy} approach and consider how PPR scoring affects player values."""
@server.prompt
def matchup_analysis(team_a: str, team_b: str, week: int) -> str:
"""Generate a prompt for head-to-head matchup analysis."""
return f"""Analyze the fantasy football matchup between {team_a} and {team_b} for Week {week}.
Compare:
1. Starting lineup projections
2. Key positional advantages
3. Weather/venue factors
4. Recent performance trends
5. Injury reports and player status
6. Predicted outcome and confidence level
Provide a detailed breakdown with specific player recommendations."""
@server.prompt
def waiver_wire_priority(league_key: str, position: str, budget: int) -> str:
"""Generate a prompt for waiver wire priority recommendations."""
return f"""Analyze waiver wire options for {position} in league {league_key} with a budget of ${budget}.
Evaluate:
1. Top 5 available players at {position}
2. FAAB bid recommendations
3. Long-term vs short-term value
4. Injury replacements vs upgrades
5. Schedule analysis for upcoming weeks
Prioritize based on immediate need and future potential."""
@server.prompt
def trade_evaluation(team_a: str, team_b: str, proposed_trade: str) -> str:
"""Generate a prompt for trade evaluation."""
return f"""Evaluate this fantasy football trade proposal between {team_a} and {team_b}:
Proposed Trade: {proposed_trade}
Analyze:
1. Fairness and value balance
2. Team needs and fit
3. Positional scarcity impact
4. Playoff schedule implications
5. Risk vs reward assessment
6. Alternative trade suggestions
Provide a clear recommendation with reasoning."""
@server.prompt
def start_sit_decision(league_key: str, position: str, player_names: list[str], week: int) -> str:
"""Generate a prompt for start/sit decision making."""
players_str = ", ".join(player_names)
return f"""Help me decide who to START at {position} for Week {week} in league {league_key}.
Players to consider: {players_str}
Analyze:
1. Projected points and ceiling/floor
2. Matchup quality and defensive rankings
3. Recent performance trends (last 3 weeks)
4. Injury concerns and game status
5. Weather and game environment factors
6. Target share / snap count / usage trends
7. Game script prediction (positive/negative)
Provide a clear START/SIT recommendation with confidence level and reasoning."""
@server.prompt
def bye_week_planning(league_key: str, team_key: str, upcoming_weeks: int) -> str:
"""Generate a prompt for bye week planning and roster management."""
return f"""Plan for upcoming bye weeks for team {team_key} in league {league_key} over the next {upcoming_weeks} weeks.
Analyze:
1. Which starters have byes in each week
2. Current bench depth at affected positions
3. Waiver wire options to cover gaps
4. Potential streaming candidates
5. Drop candidates to make room
6. Multi-week planning strategy
Provide a week-by-week action plan."""
@server.prompt
def playoff_preparation(league_key: str, team_key: str, current_week: int) -> str:
"""Generate a prompt for playoff preparation strategy."""
return f"""Create a playoff preparation strategy for team {team_key} in league {league_key} (currently Week {current_week}).
Focus on:
1. Playoff schedule strength analysis (Weeks 15-17)
2. Key players to acquire before deadline
3. Handcuffs and insurance plays
4. Bench streamlining for playoff roster
5. Injury risk assessment for key players
6. Championship-winning moves to make now
7. Weather considerations for late season
Provide actionable recommendations to maximize playoff success."""
@server.prompt
def trade_proposal_generation(league_key: str, my_team_key: str, target_team_key: str, position_need: str) -> str:
"""Generate a prompt for creating fair trade proposals."""
return f"""Generate fair trade proposals between my team ({my_team_key}) and {target_team_key} in league {league_key}.
My need: {position_need}
Create proposals that:
1. Address my positional need
2. Fill a gap for the other team
3. Are fair value for both sides
4. Consider team contexts and records
5. Account for bye weeks and playoffs
6. Include 2-3 different trade options
For each proposal explain why it works for both teams."""
@server.prompt
def injury_replacement_strategy(league_key: str, injured_player: str, injury_length: str, position: str) -> str:
"""Generate a prompt for injury replacement analysis."""
return f"""My player {injured_player} ({position}) is injured for approximately {injury_length} in league {league_key}.
Develop a replacement strategy:
1. Short-term vs long-term replacement approach
2. Top 5 waiver wire targets with analysis
3. Trade targets if waiver wire is thin
4. FAAB bidding strategy (if applicable)
5. Handcuff analysis for the injured player's backup
6. Roster moves needed (drops to consider)
7. Timeline for return and stash strategy
Provide immediate action items and contingency plans."""
@server.prompt
def streaming_dst_kicker(league_key: str, week: int, position: str) -> str:
"""Generate a prompt for streaming defense or kicker recommendations."""
pos_full = "Defense/Special Teams" if position == "DEF" else "Kicker"
return f"""Recommend {pos_full} streaming options for Week {week} in league {league_key}.
Analyze:
1. Top 5 available {pos_full} options this week
2. Matchup analysis and opponent rankings
3. Vegas lines and game environment
4. Weather factors (if relevant)
5. Next 2-3 weeks schedule preview
6. Season-long hold vs weekly stream
7. Ownership percentage and availability
Rank options with confidence levels and reasoning."""
@server.prompt
def season_long_strategy_check(league_key: str, team_key: str, current_record: str, weeks_remaining: int) -> str:
"""Generate a prompt for comprehensive season strategy assessment."""
return f"""Assess season-long strategy for team {team_key} in league {league_key}.
Current record: {current_record}
Weeks remaining: {weeks_remaining}
Comprehensive analysis:
1. Playoff probability and path
2. Win-now vs build-for-future approach
3. Trade deadline strategy (aggressive/hold/sell)
4. Waiver wire priority adjustments
5. Key matchups and must-win games
6. Positional advantages vs league
7. Risk tolerance recommendations
Provide strategic guidance for rest of season."""
@server.prompt
def weekly_game_plan(league_key: str, team_key: str, opponent_team_key: str, week: int) -> str:
"""Generate a comprehensive weekly game plan prompt."""
return f"""Create a complete game plan for Week {week} matchup between {team_key} and {opponent_team_key} in league {league_key}.
Develop strategy covering:
1. Optimal starting lineup with justification
2. Start/sit decisions with reasoning
3. Opponent's likely lineup and key players
4. Positional advantages to exploit
5. Risk assessment (safe plays vs boom/bust)
6. Weather and game environment factors
7. Waiver claims needed before games
8. Expected score and win probability
Provide a complete action plan for maximum points."""
# ============================================================================
# RESOURCES - Static and dynamic data for LLM context
# ============================================================================
@server.resource("config://scoring")
def get_scoring_rules() -> str:
"""Provide standard fantasy football scoring rules for context."""
return """Fantasy Football Scoring Rules:
PASSING:
- Passing TD: 4 points
- Passing Yards: 1 point per 25 yards
- Interception: -2 points
- 2-Point Conversion: 2 points
RUSHING:
- Rushing TD: 6 points
- Rushing Yards: 1 point per 10 yards
- 2-Point Conversion: 2 points
RECEIVING:
- Receiving TD: 6 points
- Receiving Yards: 1 point per 10 yards
- Reception: 1 point (PPR - Points Per Reception)
- 2-Point Conversion: 2 points
KICKING:
- Field Goal 0-39 yards: 3 points
- Field Goal 40-49 yards: 4 points
- Field Goal 50+ yards: 5 points
- Extra Point: 1 point
DEFENSE/SPECIAL TEAMS:
- Touchdown: 6 points
- Safety: 2 points
- Interception: 2 points
- Fumble Recovery: 2 points
- Sack: 1 point
- Blocked Kick: 2 points
- Points Allowed 0: 10 points
- Points Allowed 1-6: 7 points
- Points Allowed 7-13: 4 points
- Points Allowed 14-20: 1 point
- Points Allowed 21-27: 0 points
- Points Allowed 28-34: -1 point
- Points Allowed 35+: -4 points
SCORING VARIATIONS:
- Standard (Non-PPR): 0 points per reception
- Half-PPR: 0.5 points per reception
- Full-PPR: 1 point per reception (most common)
- Super-PPR: 1.5+ points per reception
PPR IMPACT:
- Increases value of pass-catching RBs and slot WRs
- Makes WRs more valuable relative to RBs
- Favors high-volume receivers over big-play specialists
- Changes draft strategy and player rankings"""
@server.resource("config://positions")
def get_position_info() -> str:
"""Provide fantasy football position information and requirements."""
return """Fantasy Football Position Requirements:
STANDARD LEAGUE (10-12 teams):
- QB: 1 starter
- RB: 2 starters
- WR: 2 starters
- TE: 1 starter
- FLEX: 1 (RB/WR/TE)
- K: 1 starter
- DEF/ST: 1 starter
- Bench: 6-7 players
SUPERFLEX LEAGUE:
- QB: 1 starter
- RB: 2 starters
- WR: 2 starters
- TE: 1 starter
- FLEX: 1 (RB/WR/TE)
- SUPERFLEX: 1 (QB/RB/WR/TE)
- K: 1 starter
- DEF/ST: 1 starter
- Bench: 6-7 players
POSITION ABBREVIATIONS:
- QB: Quarterback
- RB: Running Back
- WR: Wide Receiver
- TE: Tight End
- K: Kicker
- DEF/ST: Defense/Special Teams
- FLEX: Flexible position (RB/WR/TE)
- SUPERFLEX: Super flexible position (QB/RB/WR/TE)"""
@server.resource("config://strategies")
def get_draft_strategies() -> str:
"""Provide information about different fantasy football draft strategies."""
return """Fantasy Football Draft Strategies:
CONSERVATIVE STRATEGY:
- Focus on safe, high-floor players
- Prioritize proven veterans
- Avoid injury-prone players
- Build depth over upside
- Target consistent performers
- Good for beginners
BALANCED STRATEGY:
- Mix of safe picks and upside plays
- Balance risk and reward
- Target value at each pick
- Consider positional scarcity
- Adapt to draft flow
- Most popular approach
AGGRESSIVE STRATEGY:
- Target high-upside players
- Take calculated risks
- Focus on ceiling over floor
- Target breakout candidates
- Embrace volatility
- High risk, high reward
POSITIONAL STRATEGIES:
- Zero RB: Wait on running backs (more viable in PPR)
- Hero RB: Draft one elite RB early
- Robust RB: Load up on running backs
- Late Round QB: Wait on quarterback
- Streaming: Target favorable matchups
PPR-SPECIFIC STRATEGIES:
- Target pass-catching RBs (higher floor in PPR)
- Prioritize high-volume WRs over big-play specialists
- Consider slot receivers and possession WRs
- Elite TEs become more valuable (reception floor)
- RB handcuffs less critical (more WR depth)
KEY PRINCIPLES:
- Value-based drafting
- Positional scarcity awareness
- Handcuff important players
- Monitor bye weeks
- Stay flexible and adapt
- PPR changes player values significantly"""
@server.resource("data://injury-status")
def get_injury_status_info() -> str:
"""Provide information about fantasy football injury statuses."""
return """Fantasy Football Injury Status Guide:
QUESTIONABLE (Q):
- 50% chance to play
- Monitor closely
- Have backup ready
- Check game-time decisions
DOUBTFUL (D):
- 25% chance to play
- Likely to sit out
- Start backup if available
- High risk to start
OUT (O):
- Will not play
- Do not start
- Use backup or waiver pickup
- Check IR eligibility
PROBABLE (P):
- 75% chance to play
- Likely to start
- Monitor for changes
- Generally safe to start
INJURED RESERVE (IR):
- Out for extended time
- Can be stashed in IR slot
- Check league rules
- Monitor return timeline
COVID-19:
- Follow league protocols
- Check testing status
- Monitor updates
- Have backup plans
INACTIVE:
- Will not play
- Game-day decision
- Use alternative options
- Check pre-game reports"""
@server.resource("guide://weekly-strategy")
def get_weekly_strategy_guide() -> str:
"""Provide week-by-week fantasy football strategic guidance."""
return """Fantasy Football Weekly Strategy Guide:
WEEKS 1-4 (EARLY SEASON):
- Trust preseason rankings and projections
- Don't overreact to single-game performances
- Monitor snap counts and target shares
- Identify emerging trends early
- Stock up on high-upside bench stashes
- Be aggressive on waiver wire for breakouts
- Avoid panic trades after Week 1
WEEKS 5-8 (MID-SEASON):
- Sample size now meaningful for trends
- Target buy-low candidates after slow starts
- Sell high on overperformers
- Plan ahead for bye week hell
- Consolidate depth via 2-for-1 trades
- Stream defenses based on matchups
- Monitor injury reports closely
WEEKS 9-12 (PLAYOFF PUSH):
- Focus on playoff schedule (Weeks 15-17)
- Trade deadline strategy crucial
- Handcuff your stud RBs
- Drop low-floor bench players
- Target players returning from injury
- Win-now moves for playoff teams
- Sell future value if competing
WEEKS 13-14 (PLAYOFF PREP):
- Lock in your playoff roster
- Drop underperformers without hesitation
- Stream defenses for playoff weeks
- Stash handcuffs for key players
- Monitor weather for late season games
- Rest concerns for locked playoff teams
- Final waiver wire pickups
WEEKS 15-17 (PLAYOFFS):
- Championship mentality
- Weather is critical factor
- Monitor resting starters in Week 17
- Have backup plans for all positions
- Trust your studs in playoffs
- Avoid cute plays and overthinking
- Weather-proof your lineup if possible
KEY WEEKLY TASKS:
1. Check injury reports (Wed/Thu/Fri)
2. Review snap counts and usage from prior week
3. Analyze upcoming matchups
4. Submit waiver claims (Tuesday/Wednesday)
5. Check starting lineup before games
6. Monitor weather reports (Saturday/Sunday)
7. Set backup plans for questionable players"""
@server.resource("guide://common-mistakes")
def get_common_mistakes_guide() -> str:
"""Provide guidance on common fantasy football mistakes to avoid."""
return """Common Fantasy Football Mistakes to Avoid:
DRAFT MISTAKES:
❌ Drafting based on team loyalty
❌ Ignoring bye weeks completely
❌ Reaching for your favorite players
❌ Not adjusting to league scoring
❌ Following outdated rankings
❌ Drafting kicker/defense too early
❌ Ignoring injury history
✅ Value-based drafting with flexibility
✅ Balance safety and upside
✅ Adjust for PPR vs Standard scoring
IN-SEASON MISTAKES:
❌ Overreacting to one bad game
❌ Starting players on bye week
❌ Ignoring weather conditions
❌ Holding too many QBs/TEs/Defenses
❌ Not using all roster spots
❌ Forgetting to set lineup
❌ Trading based on emotion
✅ Use data and trends for decisions
✅ Stay active on waiver wire
✅ Make roster moves every week
WAIVER WIRE MISTAKES:
❌ Burning #1 priority too early
❌ Missing Wednesday waivers
❌ Not checking injury reports
❌ Chasing last week's points
❌ Ignoring opportunity (volume > talent early)
❌ Dropping players after one bad game
✅ Target volume and opportunity
✅ Plan ahead for bye weeks
✅ Be patient with waiver priority
TRADE MISTAKES:
❌ Accepting first offer received
❌ Trading based on name value only
❌ Ignoring team context and situation
❌ Not considering playoff schedule
❌ Vetoing trades out of spite
❌ Trading away depth before bye weeks
❌ Panicking after injuries
✅ Always counter-offer first
✅ Consider both teams' needs
✅ Look at rest-of-season schedules
LINEUP MISTAKES:
❌ Benching studs after bad game
❌ Starting players on snap count
❌ Overthinking Thursday night games
❌ Not checking start times
❌ Ignoring weather reports
❌ Starting questionable players without backup
❌ Getting too cute with lineup
✅ Start your studs
✅ Have contingency plans
✅ Trust projections over gut
STRATEGIC MISTAKES:
❌ Playing for second place
❌ Not taking calculated risks
❌ Holding players for trade value
❌ Ignoring playoff implications
❌ Not handcuffing elite RBs
❌ Hoarding too many bench RBs
✅ Championship-or-bust mentality
✅ Maximize every roster spot
✅ Make bold moves when necessary"""
@server.resource("guide://advanced-stats")
def get_advanced_stats_glossary() -> str:
"""Provide glossary of advanced fantasy football statistics."""
return """Advanced Fantasy Football Statistics Glossary:
VOLUME METRICS:
- Snap Count %: Percentage of offensive snaps played
→ 70%+ is ideal for RB/WR, 90%+ for elite
- Target Share: Percentage of team targets received
→ 20%+ is WR1 territory, 25%+ is elite
- Touch Count: Total rushing attempts + receptions
→ 15+ touches for RB1, 20+ is workhorse territory
- Red Zone Touches: Carries/targets inside opponent 20
→ High correlation with TDs and fantasy points
- Air Yards: Total depth of targets (catchable or not)
→ Higher air yards = more big play potential
EFFICIENCY METRICS:
- Yards Per Route Run (YPRR): Receiving yards per route
→ 2.0+ is excellent, 2.5+ is elite
- Yards After Contact (YAC): Rushing/receiving yards after contact
→ Indicates home run ability and toughness
- Yards Per Carry (YPC): Rushing efficiency
→ 4.5+ is good, 5.0+ is excellent
- True Catch Rate: Catchable targets caught
→ Better than raw catch % for WR evaluation
- Broken Tackles: Missed tackles forced
→ Indicates elusiveness and big play ability
SITUATION METRICS:
- Game Script: Expected point differential
→ Positive = more passing, Negative = more rushing
- Neutral Game Script %: Snaps in neutral situations
→ Better indicator of true role than blowouts
- Two-Minute Drill Usage: Involvement in hurry-up
→ Indicates trust and pass-catching ability
- Goal Line Carries: Touches inside 5-yard line
→ TD equity indicator for RBs
OPPORTUNITY METRICS:
- Expected Fantasy Points (xFP): Based on usage
→ Compare actual vs expected to find efficiency
- Opportunity Share: Team offense share
→ Volume is king in fantasy football
- Slot Rate: % of snaps in slot for WRs
→ Slot WRs see more targets in PPR
- Route Participation: % of pass plays running route
→ 90%+ indicates featured receiver
QUARTERBACK METRICS:
- Time to Throw: Average release time
→ Affects WR separation and completion %
- Play Action %: % of dropbacks using play action
→ Higher = more big plays downfield
- Pressure Rate: % of dropbacks under pressure
→ Affects turnovers and efficiency
- Deep Ball %: % of throws 20+ yards
→ Indicates downfield aggression
SKILL POSITION TRENDS:
- Trending Up: Increased snap %, target share, touches
- Trending Down: Decreased involvement or efficiency
- Consistent: Stable role week-to-week
- Volatile: Boom/bust performances
KEY TAKEAWAYS:
→ Volume > Talent in fantasy (especially early season)
→ Opportunity + Role > Efficiency alone
→ Target RBs with 15+ touches and WRs with 20%+ target share
→ Red zone usage is most predictive of TDs
→ Monitor snap counts for emerging players"""
@server.resource("guide://playoff-strategies")
def get_playoff_strategies() -> str:
"""Provide strategies for fantasy football playoffs."""
return """Fantasy Football Playoff Strategies:
ROSTER CONSTRUCTION FOR PLAYOFFS:
✓ Handcuff elite RBs (injury insurance)
✓ Drop low-floor bench players
✓ Prioritize favorable playoff schedules (Weeks 15-17)
✓ Stream defense matchups
✓ Have backup plans for every position
✓ Consolidate depth via trades before deadline
✓ Target players returning from injury
PRE-PLAYOFF PREPARATION (Weeks 12-14):
1. Analyze Week 15-17 schedules for all players
2. Identify teams likely to rest starters (Week 17)
3. Target defenses playing poor offenses in playoffs
4. Trade away future value for immediate upgrades
5. Prioritize players on pass-heavy offenses
6. Stock handcuffs for your RB1/RB2
7. Drop players on bye in Week 14
CHAMPIONSHIP WEEK STRATEGY (Week 16-17):
- Weather is critical (snow/wind affects passing)
- Monitor news for resting starters
- Indoor games safer than outdoor in December
- Volume over talent for borderline decisions
- Trust proven performers over hot waiver adds
- Have Saturday replacements for Sunday players
- Check Vegas lines (blowouts = less volume for studs)
PLAYOFF SCHEDULE ANALYSIS:
GOOD PLAYOFF MATCHUPS (Target):
- Bad pass defenses (allows 250+ pass yards/game)
- Bad run defenses (allows 130+ rush yards/game)
- High-scoring offenses (creates game script)
- Dome games in late December (weather-proof)
- Teams eliminated from playoffs (less effort)
BAD PLAYOFF MATCHUPS (Avoid):
- Elite defenses (top 5 in points allowed)
- Divisional revenge games (extra motivation)
- Cold weather games for warm weather teams
- Week 17 locked playoff seeds (rest risk)
- Backup QBs or depleted offenses
POSITIONAL STRATEGY:
QUARTERBACK:
- Target high-volume passers (35+ attempts)
- Prefer indoor or warm-weather games
- Avoid QBs on run-heavy teams in playoffs
- Stream based on matchup if no elite option
RUNNING BACK:
- Handcuff all workhorse RBs
- Target RBs with bellcow usage (20+ touches)
- Avoid RBBC situations in playoffs
- Monitor for rest in Week 17 for playoff teams
- Prefer pass-catching backs in PPR
WIDE RECEIVER:
- Target high-volume WRs (8+ targets)
- Slot receivers safer in bad weather
- Deep threats risky in wind/snow
- WR1s on team safer than WR2/3
- Avoid rookie QBs throwing in bad weather
TIGHT END:
- Elite TEs (Kelce tier) are matchup-proof
- Stream TEs against bad defenses otherwise
- Red zone usage critical for TE scoring
- Volume matters more than talent
FLEX DECISIONS:
- Prefer RBs over WRs in bad weather
- WRs have higher ceiling in good matchups
- TEs are floor plays (safe but low ceiling)
- Trust your studs over waiver wire adds
- Volume > Matchup for borderline decisions
DEFENSE/KICKER STREAMING:
- Stream defense vs bad offenses
- Target defenses at home in bad weather
- Kickers in domes for consistency
- Avoid defenses vs elite QBs
WEEK 17 CONSIDERATIONS:
⚠️ Teams with locked playoff seeds may rest starters
⚠️ Monitor Saturday injury reports closely
⚠️ Have backup plans for every starter
⚠️ Avoid players on locked 1-seed teams
⚠️ Target teams fighting for playoff spots
CHAMPIONSHIP MENTALITY:
💪 Trust the players who got you here
💪 Don't overthink lineup decisions
💪 Weather and game script matter most
💪 Volume and opportunity = floor
💪 Have contingency plans ready
💪 Championship = bold moves + smart process"""
@server.resource("guide://dynasty-keeper")
def get_dynasty_keeper_guide() -> str:
"""Provide strategies for dynasty and keeper leagues."""
return """Dynasty & Keeper League Strategy Guide:
DYNASTY LEAGUE FUNDAMENTALS:
- Player values span multiple years
- Youth and upside trump proven veterans
- Draft picks are valuable trade assets
- Rebuild vs compete decisions critical
- Contracts and cap space management (if applicable)
- Deeper benches (25-30+ roster spots typical)
KEEPER LEAGUE FUNDAMENTALS:
- Keep 1-5 players year-to-year (league dependent)
- Keeper cost tied to draft position or auction $
- Balance current year vs future value
- Late-round picks provide keeper value
- Drop players with bad keeper value late season
VALUATION DIFFERENCES (Dynasty vs Redraft):
POSITIONS TO PRIORITIZE:
1. Elite Young RBs (age 22-25)
→ Rare asset with multi-year value
2. Young WRs with target share (age 22-27)
→ Longer careers than RBs, safer dynasty assets
3. Young elite TEs (age 23-26)
→ Kelce/Andrews tier, decade-long value
4. Top 5 QBs in Superflex
→ Game-breaking advantage in Superflex formats
ROOKIE DRAFT STRATEGY:
- Early picks = high-capital NFL draft picks
- Target landing spot + draft capital combination
- RBs have shorter shelf life but immediate impact
- WRs take 2-3 years to develop typically
- QBs in Superflex leagues = premium value
- Avoid reaching for need (value > need in dynasty)
PLAYER LIFECYCLE MANAGEMENT:
CONTENDING TEAMS (Win Now):
→ Trade future picks for proven vets
→ Target players aged 26-29 (prime years)
→ Package young players for upgrades
→ Stream and optimize for current season
→ Don't hold onto taxi squad guys
REBUILDING TEAMS (2+ Years Out):
→ Trade aging vets for picks
→ Acquire young players with upside
→ Take on injured players for discount
→ Don't compete half-way (commit to rebuild)
→ Accumulate draft capital (1sts and 2nds)
AGING CURVE BY POSITION:
- RB: Peak age 24-27, cliff at 28-30
- WR: Peak age 25-29, productive to 32+
- TE: Peak age 25-30, productive to 33+
- QB: Peak age 27-35, can play to 40+
TRADE STRATEGY:
SELLING WINDOW (Trade Before Value Drops):
- RBs aged 28+ (especially with injuries)
- WRs aged 31+ (target win-now teams)
- Players on contract years (uncertainty)
- Boom/bust players after hot streak
- Backup RBs before starter returns
BUYING WINDOW (Acquire at Discount):
- Injured players from contenders
- Rookies after slow start (patience pays)
- Players in bad offenses (situation change)
- Young WRs breaking out (buy early)
- Players on new teams (positive change)
DRAFT PICK VALUES:
- 1st Round Picks: Premium assets (especially early)
- 2nd Round Picks: Solid value, trade fodder
- 3rd+ Round Picks: Dart throws, low hit rate
TYPICAL PICK VALUE (Dynasty):
- Early 1st (1.01-1.03): Established WR2/RB2
- Mid 1st (1.04-1.08): Young WR2 or aging RB1
- Late 1st (1.09-1.12): WR3 with upside or TE1
- Early 2nd: High-upside WR or backup RB
- Mid/Late 2nd: Bench depth or taxi squad stash
KEEPER LEAGUE SPECIFIC:
KEEPER VALUE CALCULATION:
- Keep cost vs Expected draft position
- Years of keeper eligibility remaining
- Contract escalation (if applicable)
- Opportunity cost of keeper slot
BEST KEEPER VALUES:
✓ Late round picks who broke out (round 10+ keepers)
✓ Rookies drafted late who hit (league-winning value)
✓ Injured players stashed (return to form)
✓ Young QBs in Superflex (early breakouts)
AVOID KEEPING:
✗ Early round picks (no value gain)
✗ Aging RBs (value cliff coming)
✗ Players with bad contracts (auction leagues)
✗ Injury-prone vets (risk > reward)
KEY DIFFERENCES VS REDRAFT:
📊 Think 2-3 years ahead, not just this season
📊 Age matters more than current production
📊 Target situation + talent over production only
📊 Rebuild fully or compete fully (no half-measures)
📊 Draft picks are tradeable assets with real value
📊 Patience is rewarded (develop young players)
📊 Deeper benches = more roster management"""
def run_http_server(
host: Optional[str] = None, port: Optional[int] = None, *, show_banner: bool = True
) -> None:
"""Start the FastMCP server using the HTTP transport."""
resolved_host = host or os.getenv("HOST", "0.0.0.0")
resolved_port = port or int(os.getenv("PORT", "8000"))
server.run(
"http",
host=resolved_host,
port=resolved_port,
show_banner=show_banner,
)
def main() -> None:
"""Console script entry point for launching the HTTP server."""
run_http_server()
__all__ = [
"server",
"run_http_server",
"main",
# Core Tools
"ff_get_leagues",
"ff_get_league_info",
"ff_get_standings",
"ff_get_roster",
"ff_get_matchup",
"ff_get_players",
"ff_compare_teams",
"ff_build_lineup",
"ff_refresh_token",
"ff_get_api_status",
"ff_clear_cache",
"ff_get_draft_results",
"ff_get_waiver_wire",
"ff_get_draft_rankings",
"ff_get_draft_recommendation",
"ff_analyze_draft_state",
"ff_analyze_reddit_sentiment",
# Prompts - Pre-built prompt templates for LLMs
"analyze_roster_strengths",
"draft_strategy_advice",
"matchup_analysis",
"waiver_wire_priority",
"trade_evaluation",
"start_sit_decision",
"bye_week_planning",
"playoff_preparation",
"trade_proposal_generation",
"injury_replacement_strategy",
"streaming_dst_kicker",
"season_long_strategy_check",
"weekly_game_plan",
# Resources - Reference data for LLM context
"get_scoring_rules",
"get_position_info",
"get_draft_strategies",
"get_injury_status_info",
"get_weekly_strategy_guide",
"get_common_mistakes_guide",
"get_advanced_stats_glossary",
"get_playoff_strategies",
"get_dynasty_keeper_guide",
"get_tool_selection_guide",
"get_version",
]
# Optional resource: expose deployed commit SHA for diagnostics
try:
with open(os.path.join(os.path.dirname(__file__), "COMMIT_SHA"), "r", encoding="utf-8") as _f:
_COMMIT_SHA = _f.read().strip()
except Exception: # pragma: no cover - best effort
_COMMIT_SHA = "unknown"
@server.resource("guide://tool-selection")
def get_tool_selection_guide() -> str:
"""Comprehensive guide for LLMs on when and how to use fantasy football tools."""
return json.dumps(
{
"title": "Fantasy Football Tool Selection Guide for LLMs",
"description": "Strategic guidance for AI assistants on optimal tool usage patterns",
"workflow_priority": [
"1. START: ff_get_leagues - Always begin here if you don't have a league_key",
"2. CONTEXT: ff_get_league_info - Understand league settings and scoring",
"3. BASELINE: ff_get_roster - Know current lineup before making recommendations",
"4. COMPETITION: ff_get_matchup - Analyze weekly opponent for strategic adjustments",
"5. OPPORTUNITIES: ff_get_waiver_wire - Identify available upgrades",
"6. OPTIMIZATION: ff_build_lineup - AI-powered lineup construction",
],
"tool_categories": {
"CORE_LEAGUE_DATA": {
"description": "Essential league information and setup",
"tools": {
"ff_get_leagues": "Discovery: Find available leagues and extract league_key identifiers",
"ff_get_league_info": "Configuration: League settings, scoring rules, roster requirements",
"ff_get_standings": "Rankings: Current standings, records, points for strategy context",
},
},
"PLAYER_ROSTER_ANALYSIS": {
"description": "Player and roster management tools",
"tools": {
"ff_get_roster": "Current Lineup: Configurable roster data (basic/standard/full detail levels) for lineup decisions",
"ff_get_players": "Player Search: Find specific players by name or position",
"ff_get_waiver_wire": "Free Agents: Available players with advanced metrics",
},
},
"MATCHUP_COMPETITION": {
"description": "Head-to-head analysis and competitive intelligence",
"tools": {
"ff_get_matchup": "Opponent Analysis: Weekly head-to-head strategic insights",
"ff_compare_teams": "Team Comparison: Direct roster and performance comparisons",
},
},
"OPTIMIZATION_STRATEGY": {
"description": "AI-powered decision making and strategy tools",
"tools": {
"ff_build_lineup": "AI Optimization: Championship-level lineup recommendations with positional constraints",
"ff_get_draft_rankings": "Player Tiers: Value assessment and tier-based rankings",
"ff_analyze_reddit_sentiment": "Market Intelligence: Public opinion and trending players",
},
},
"ADVANCED_ANALYSIS": {
"description": "Deep analytics and historical insights",
"tools": {
"ff_get_draft_results": "Draft History: Historical patterns and team building analysis",
"ff_analyze_draft_state": "Live Draft: Real-time draft strategy and recommendations",
},
},
"UTILITY_MAINTENANCE": {
"description": "System maintenance and troubleshooting",
"tools": {
"ff_refresh_token": "Authentication: Fix Yahoo API authentication issues",
"ff_get_api_status": "Health Check: Verify system status and connectivity",
"ff_clear_cache": "Reset: Clear cached data for fresh analysis",
},
},
},
"strategic_usage_patterns": {
"weekly_lineup_optimization": [
"ff_get_leagues -> ff_get_roster -> ff_get_matchup -> ff_get_waiver_wire -> ff_build_lineup"
],
"draft_preparation": [
"ff_get_leagues -> ff_get_league_info -> ff_get_draft_rankings -> ff_analyze_draft_state"
],
"competitive_analysis": [
"ff_get_league_info -> ff_get_standings -> ff_compare_teams -> ff_get_matchup"
],
"market_research": [
"ff_get_waiver_wire -> ff_analyze_reddit_sentiment -> ff_get_players"
],
},
"decision_framework": {
"data_gathering": "Always start with league discovery and current roster state",
"context_building": "Understand league settings, scoring, and competitive landscape",
"opportunity_identification": "Use waiver wire and sentiment analysis for edge cases",
"optimization": "Apply AI-powered tools for championship-level recommendations",
"validation": "Cross-reference multiple data sources for confident decisions",
},
"best_practices": [
"NEVER guess league_key - always use ff_get_leagues first",
"ALWAYS check current roster before making lineup recommendations",
"USE ff_get_matchup for opponent-specific weekly strategy",
"LEVERAGE ff_analyze_reddit_sentiment for contrarian plays",
"APPLY strategy parameters in ff_build_lineup for optimized construction",
"COMBINE multiple tools for comprehensive decision making",
],
}
)
@server.resource("meta://version")
def get_version() -> str: # pragma: no cover - simple accessor
return json.dumps({"commit": _COMMIT_SHA})
if __name__ == "__main__":
main()
```