Capstone - Complete Analytics System

# Streamlit equivalent structure
"""
Save as app.py and run with: streamlit run app.py
"""

import streamlit as st
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go

# Sample data
player_data = pd.DataFrame({
    "player": ["Haaland", "Salah", "Kane", "Saka", "Rashford"],
    "goals": [36, 19, 30, 14, 17],
    "xG": [28.5, 20.2, 25.8, 12.1, 15.8],
    "assists": [8, 12, 3, 11, 5],
    "xA": [5.2, 9.8, 4.5, 8.2, 6.1]
})

# Page config
st.set_page_config(page_title="Player Analytics", layout="wide")
st.title("Premier League Player Analytics")

# Sidebar controls
st.sidebar.header("Filters")
selected_player = st.sidebar.selectbox("Select Player", player_data["player"])
metric = st.sidebar.selectbox("Primary Metric", ["goals", "xG", "assists"])
min_goals = st.sidebar.slider("Minimum Goals", 0, 40, 10)

# Filter data
filtered_data = player_data[player_data["goals"] >= min_goals]

# Main content - tabs
tab1, tab2, tab3 = st.tabs(["Overview", "Player Profile", "Comparison"])

with tab1:
    col1, col2 = st.columns(2)

    with col1:
        # Scatter plot
        fig = px.scatter(filtered_data, x="xG", y="goals",
                        text="player", title="Goals vs Expected Goals")
        fig.add_shape(type="line", x0=0, x1=40, y0=0, y1=40,
                     line=dict(dash="dash", color="red"))
        st.plotly_chart(fig, use_container_width=True)

    with col2:
        # Summary table
        summary = filtered_data.copy()
        summary["overperformance"] = summary["goals"] - summary["xG"]
        summary["efficiency"] = (summary["goals"] / summary["xG"]).round(2)
        st.dataframe(summary.sort_values(metric, ascending=False))

with tab2:
    player = player_data[player_data["player"] == selected_player].iloc[0]

    st.subheader(f"Player Profile: {selected_player}")

    col1, col2, col3, col4 = st.columns(4)
    col1.metric("Goals", player["goals"], f"xG: {player['xG']}")
    col2.metric("Assists", player["assists"], f"xA: {player['xA']}")
    col3.metric("Overperformance", round(player["goals"] - player["xG"], 1))
    col4.metric("Efficiency", f"{(player['goals']/player['xG']):.2f}")

with tab3:
    # Comparison bar chart
    fig = go.Figure(data=[
        go.Bar(name="Goals", x=filtered_data["player"], y=filtered_data["goals"],
               marker_color="#1B5E20"),
        go.Bar(name="xG", x=filtered_data["player"], y=filtered_data["xG"],
               marker_color="#FFD700")
    ])
    fig.update_layout(barmode="group", title="Goals vs xG Comparison")
    st.plotly_chart(fig, use_container_width=True)

# FastAPI Implementation
"""
Save as main.py and run with: uvicorn main:app --reload
"""

from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel
from typing import List, Optional
import pandas as pd

app = FastAPI(
    title="Football Analytics API",
    description="API for football player statistics",
    version="1.0.0"
)

# Sample data (in production, use a database)
player_db = pd.DataFrame({
    "id": [1, 2, 3, 4, 5],
    "name": ["Haaland", "Salah", "Kane", "Saka", "Rashford"],
    "team": ["Man City", "Liverpool", "Bayern", "Arsenal", "Man Utd"],
    "goals": [36, 19, 30, 14, 17],
    "xG": [28.5, 20.2, 25.8, 12.1, 15.8],
    "position": ["ST", "RW", "ST", "RW", "LW"]
})

# Pydantic models for request/response
class Player(BaseModel):
    id: int
    name: str
    team: str
    goals: int
    xG: float
    position: str

class PlayerScore(BaseModel):
    player: str
    score: float
    weights_used: dict

class LeagueSummary(BaseModel):
    total_players: int
    avg_goals: float
    avg_xG: float
    top_scorer: str

# Endpoints
@app.get("/players", response_model=List[Player])
async def get_all_players(
    position: Optional[str] = Query(None, description="Filter by position"),
    min_goals: int = Query(0, description="Minimum goals")
):
    """Get all players with optional filters"""
    result = player_db.copy()

    if position:
        result = result[result["position"] == position]
    result = result[result["goals"] >= min_goals]

    return result.to_dict("records")

@app.get("/players/{player_id}", response_model=Player)
async def get_player(player_id: int):
    """Get player by ID"""
    player = player_db[player_db["id"] == player_id]

    if player.empty:
        raise HTTPException(status_code=404, detail="Player not found")

    return player.iloc[0].to_dict()

@app.get("/teams/{team}/players", response_model=List[Player])
async def get_team_players(team: str):
    """Get all players for a team"""
    players = player_db[player_db["team"] == team]

    if players.empty:
        raise HTTPException(status_code=404, detail="No players found for team")

    return players.to_dict("records")

@app.post("/players/{player_id}/score", response_model=PlayerScore)
async def calculate_player_score(
    player_id: int,
    goals_weight: float = 1.0,
    xG_weight: float = 0.5
):
    """Calculate weighted player score"""
    player = player_db[player_db["id"] == player_id]

    if player.empty:
        raise HTTPException(status_code=404, detail="Player not found")

    player = player.iloc[0]
    score = player["goals"] * goals_weight + player["xG"] * xG_weight

    return PlayerScore(
        player=player["name"],
        score=score,
        weights_used={"goals": goals_weight, "xG": xG_weight}
    )

@app.get("/stats/summary", response_model=LeagueSummary)
async def get_league_summary():
    """Get league-wide statistics"""
    return LeagueSummary(
        total_players=len(player_db),
        avg_goals=player_db["goals"].mean(),
        avg_xG=player_db["xG"].mean(),
        top_scorer=player_db.loc[player_db["goals"].idxmax(), "name"]
    )

# Run with: uvicorn main:app --reload
# Access docs at: http://localhost:8000/docs

# Streamlit Dashboard with Components
"""
Professional dashboard structure in Streamlit
"""
import streamlit as st
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
from datetime import datetime

# Page configuration
st.set_page_config(
    page_title="Football Analytics Hub",
    page_icon="⚽",
    layout="wide",
    initial_sidebar_state="expanded"
)

# Custom CSS
st.markdown("""
<style>
    .metric-card {
        background-color: #1B5E20;
        color: white;
        padding: 20px;
        border-radius: 10px;
        text-align: center;
    }
    .metric-value {
        font-size: 2.5rem;
        font-weight: bold;
    }
    .metric-label {
        font-size: 0.9rem;
        opacity: 0.8;
    }
</style>
""", unsafe_allow_html=True)

# Sidebar
with st.sidebar:
    st.image("https://via.placeholder.com/150x50?text=Logo", width=150)
    st.title("Filters")

    season = st.selectbox("Season", ["2023-24", "2022-23", "2021-22"])
    league = st.selectbox("League", ["Premier League", "La Liga", "Bundesliga"])

    st.divider()
    page = st.radio("Navigation", ["Overview", "Players", "Teams", "Matches"])

# Sample data
@st.cache_data
def load_data():
    return pd.DataFrame({
        "match_week": range(1, 21),
        "goals": [28, 55, 82, 110, 138, 165, 195, 222, 250, 280,
                  308, 338, 365, 395, 422, 452, 480, 510, 538, 568],
        "xG": [26, 52, 80, 105, 132, 158, 188, 215, 245, 272,
               300, 328, 358, 385, 415, 442, 472, 500, 530, 558]
    })

data = load_data()

# Main content based on page selection
if page == "Overview":
    st.title("League Overview")

    # KPI Row
    col1, col2, col3, col4 = st.columns(4)

    with col1:
        st.metric(label="Total Goals", value="568", delta="+12 vs last season")
    with col2:
        st.metric(label="Avg xG/Match", value="2.68", delta="0.15")
    with col3:
        st.metric(label="Top Scorer", value="Haaland (28)")
    with col4:
        st.metric(label="Clean Sheets", value="45", delta="-3")

    st.divider()

    # Charts row
    col1, col2 = st.columns([2, 1])

    with col1:
        st.subheader("Goals Timeline")
        fig = go.Figure()
        fig.add_trace(go.Scatter(x=data["match_week"], y=data["goals"],
                                 name="Goals", line=dict(color="#1B5E20")))
        fig.add_trace(go.Scatter(x=data["match_week"], y=data["xG"],
                                 name="xG", line=dict(color="#FFD700", dash="dash")))
        fig.update_layout(height=400)
        st.plotly_chart(fig, use_container_width=True)

    with col2:
        st.subheader("Quick Stats")
        stats_df = pd.DataFrame({
            "Metric": ["Matches Played", "Home Wins", "Away Wins", "Draws"],
            "Value": [200, 85, 65, 50]
        })
        st.dataframe(stats_df, hide_index=True, use_container_width=True)

    # Bottom row
    col1, col2 = st.columns(2)

    with col1:
        st.subheader("xG vs Actual Goals")
        scatter_data = pd.DataFrame({
            "team": ["Man City", "Arsenal", "Liverpool", "Chelsea", "Spurs"],
            "goals": [85, 78, 72, 58, 65],
            "xG": [78, 72, 70, 62, 60]
        })
        fig = px.scatter(scatter_data, x="xG", y="goals", text="team",
                        color_discrete_sequence=["#1B5E20"])
        fig.add_shape(type="line", x0=50, x1=90, y0=50, y1=90,
                     line=dict(dash="dash", color="red"))
        st.plotly_chart(fig, use_container_width=True)

    with col2:
        st.subheader("Top Performers")
        performers = pd.DataFrame({
            "Player": ["Haaland", "Salah", "Kane", "Saka", "Palmer"],
            "Goals": [28, 19, 22, 14, 16],
            "xG": [22.5, 18.2, 20.1, 12.5, 14.8],
            "Overperf": [5.5, 0.8, 1.9, 1.5, 1.2]
        })
        st.dataframe(performers, hide_index=True, use_container_width=True)

elif page == "Players":
    st.title("Player Analysis")

    # Search and filters
    col1, col2, col3 = st.columns(3)
    with col1:
        search = st.text_input("Search Player", "")
    with col2:
        position = st.selectbox("Position", ["All", "GK", "DEF", "MID", "FWD"])
    with col3:
        min_minutes = st.slider("Minimum Minutes", 0, 3000, 500)

    st.info("Player search and radar charts would appear here")

# Python Data Layer with Caching
from functools import lru_cache
from typing import Optional, List, Dict
import pandas as pd
from sqlalchemy import create_engine, text
from dataclasses import dataclass
import os
from datetime import datetime, timedelta
import redis
import json

@dataclass
class CacheConfig:
    enabled: bool = True
    ttl_seconds: int = 3600  # 1 hour

class DataManager:
    """Data access layer with caching and connection pooling"""

    def __init__(self, cache_config: Optional[CacheConfig] = None):
        self.cache_config = cache_config or CacheConfig()
        self._engine = None
        self._redis = None

    @property
    def engine(self):
        """Lazy database connection"""
        if self._engine is None:
            db_url = os.getenv("DATABASE_URL",
                              "postgresql://user:pass@localhost/football")
            self._engine = create_engine(db_url, pool_size=5, max_overflow=10)
        return self._engine

    @property
    def redis_client(self):
        """Lazy Redis connection for distributed caching"""
        if self._redis is None:
            redis_url = os.getenv("REDIS_URL", "redis://localhost:6379")
            self._redis = redis.from_url(redis_url)
        return self._redis

    def _cache_key(self, prefix: str, *args) -> str:
        """Generate cache key"""
        return f"{prefix}:{':'.join(str(a) for a in args)}"

    def _get_cached(self, key: str) -> Optional[pd.DataFrame]:
        """Get from cache"""
        if not self.cache_config.enabled:
            return None
        try:
            data = self.redis_client.get(key)
            if data:
                return pd.read_json(data)
        except:
            pass
        return None

    def _set_cached(self, key: str, df: pd.DataFrame):
        """Set cache"""
        if not self.cache_config.enabled:
            return
        try:
            self.redis_client.setex(
                key,
                self.cache_config.ttl_seconds,
                df.to_json()
            )
        except:
            pass

    def get_players(self, season: str, use_cache: bool = True) -> pd.DataFrame:
        """Get players for a season"""
        cache_key = self._cache_key("players", season)

        if use_cache:
            cached = self._get_cached(cache_key)
            if cached is not None:
                return cached

        query = text("""
            SELECT * FROM players
            WHERE season = :season
            ORDER BY goals DESC
        """)

        with self.engine.connect() as conn:
            result = pd.read_sql(query, conn, params={"season": season})

        self._set_cached(cache_key, result)
        return result

    def get_matches(self, team_id: int, season: str) -> pd.DataFrame:
        """Get matches for a team"""
        query = text("""
            SELECT m.*, t1.name as home_team, t2.name as away_team
            FROM matches m
            JOIN teams t1 ON m.home_team_id = t1.id
            JOIN teams t2 ON m.away_team_id = t2.id
            WHERE (m.home_team_id = :team_id OR m.away_team_id = :team_id)
              AND m.season = :season
            ORDER BY m.match_date
        """)

        with self.engine.connect() as conn:
            return pd.read_sql(query, conn,
                             params={"team_id": team_id, "season": season})

    def get_team_stats(self, team_id: int, season: str) -> Dict:
        """Calculate team aggregations"""
        matches = self.get_matches(team_id, season)

        is_home = matches["home_team_id"] == team_id
        wins = ((is_home & (matches["home_goals"] > matches["away_goals"])) |
                (~is_home & (matches["away_goals"] > matches["home_goals"]))).sum()

        goals_for = matches.loc[is_home, "home_goals"].sum() + \
                    matches.loc[~is_home, "away_goals"].sum()
        goals_against = matches.loc[is_home, "away_goals"].sum() + \
                        matches.loc[~is_home, "home_goals"].sum()

        return {
            "total_matches": len(matches),
            "wins": int(wins),
            "goals_for": int(goals_for),
            "goals_against": int(goals_against)
        }

    def clear_cache(self, pattern: str = "*"):
        """Clear cached data"""
        if self.cache_config.enabled:
            for key in self.redis_client.scan_iter(pattern):
                self.redis_client.delete(key)


# Usage
# dm = DataManager()
# players = dm.get_players("2023-24")
# stats = dm.get_team_stats(1, "2023-24")

# Docker deployment for Python apps
# Dockerfile

FROM python:3.11-slim

WORKDIR /app

# Install system dependencies
RUN apt-get update && apt-get install -y \
    gcc \
    libpq-dev \
    && rm -rf /var/lib/apt/lists/*

# Install Python packages
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application
COPY . .

# Expose port
EXPOSE 8501

# Healthcheck
HEALTHCHECK CMD curl --fail http://localhost:8501/_stcore/health

# Run Streamlit
CMD ["streamlit", "run", "app.py", "--server.port=8501", "--server.address=0.0.0.0"]

# ---
# requirements.txt
streamlit==1.29.0
pandas==2.1.0
plotly==5.18.0
sqlalchemy==2.0.0
psycopg2-binary==2.9.9
redis==5.0.0

# ---
# docker-compose.yml
version: "3.8"

services:
  app:
    build: .
    ports:
      - "8501:8501"
    environment:
      - DATABASE_URL=postgresql://app:${DB_PASSWORD}@db/football
      - REDIS_URL=redis://redis:6379
    depends_on:
      - db
      - redis

  db:
    image: postgres:14-alpine
    environment:
      - POSTGRES_DB=football
      - POSTGRES_USER=app
      - POSTGRES_PASSWORD=${DB_PASSWORD}
    volumes:
      - pgdata:/var/lib/postgresql/data

  redis:
    image: redis:7-alpine
    volumes:
      - redisdata:/data

volumes:
  pgdata:
  redisdata:

# Python: FastAPI with WebSockets for Live Updates
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
import asyncio
import json
from typing import List
from dataclasses import dataclass, asdict
import random
from datetime import datetime

app = FastAPI()

# Connection manager for WebSocket clients
class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []
        self.match_subscriptions: dict = {}  # match_id -> [connections]

    async def connect(self, websocket: WebSocket, match_id: int):
        await websocket.accept()
        self.active_connections.append(websocket)

        if match_id not in self.match_subscriptions:
            self.match_subscriptions[match_id] = []
        self.match_subscriptions[match_id].append(websocket)

    def disconnect(self, websocket: WebSocket, match_id: int):
        self.active_connections.remove(websocket)
        if match_id in self.match_subscriptions:
            self.match_subscriptions[match_id].remove(websocket)

    async def broadcast_match_update(self, match_id: int, data: dict):
        """Send update to all clients watching this match"""
        if match_id in self.match_subscriptions:
            for connection in self.match_subscriptions[match_id]:
                try:
                    await connection.send_json(data)
                except:
                    pass

manager = ConnectionManager()

@dataclass
class MatchEvent:
    minute: int
    event_type: str
    team: str
    player: str
    xG: float = 0.0

@dataclass
class MatchState:
    match_id: int
    minute: int
    home_xG: float
    away_xG: float
    home_goals: int
    away_goals: int
    events: List[dict]

# WebSocket endpoint for live match updates
@app.websocket("/ws/match/{match_id}")
async def websocket_match(websocket: WebSocket, match_id: int):
    await manager.connect(websocket, match_id)

    try:
        # Send initial state
        initial_state = get_match_state(match_id)
        await websocket.send_json(asdict(initial_state))

        # Keep connection alive and handle client messages
        while True:
            # Receive heartbeat or commands from client
            data = await websocket.receive_text()

            if data == "ping":
                await websocket.send_text("pong")
            elif data == "refresh":
                state = get_match_state(match_id)
                await websocket.send_json(asdict(state))

    except WebSocketDisconnect:
        manager.disconnect(websocket, match_id)

def get_match_state(match_id: int) -> MatchState:
    """Get current match state (in production, from database/API)"""
    return MatchState(
        match_id=match_id,
        minute=random.randint(1, 90),
        home_xG=round(random.uniform(0.5, 2.5), 2),
        away_xG=round(random.uniform(0.3, 2.0), 2),
        home_goals=random.randint(0, 3),
        away_goals=random.randint(0, 2),
        events=[
            {"minute": 23, "type": "Goal", "team": "Home"},
            {"minute": 45, "type": "Shot", "team": "Away"},
        ]
    )

# Background task to push updates
async def match_update_broadcaster():
    """Simulate live match updates"""
    while True:
        await asyncio.sleep(5)  # Update every 5 seconds

        for match_id in list(manager.match_subscriptions.keys()):
            if manager.match_subscriptions[match_id]:
                state = get_match_state(match_id)
                await manager.broadcast_match_update(match_id, asdict(state))

# Start background task on startup
@app.on_event("startup")
async def startup_event():
    asyncio.create_task(match_update_broadcaster())

# Streamlit with auto-refresh
"""
# app_live.py - Streamlit with auto-refresh
import streamlit as st
import time

# Auto-refresh using st.empty and time.sleep
placeholder = st.empty()

while True:
    with placeholder.container():
        # Fetch latest data
        data = fetch_live_match_data()

        st.metric("Home xG", data["home_xG"])
        st.metric("Away xG", data["away_xG"])

        # Chart updates
        st.plotly_chart(create_xg_chart(data))

    time.sleep(5)  # Refresh every 5 seconds

# Or use streamlit-autorefresh component
# pip install streamlit-autorefresh
from streamlit_autorefresh import st_autorefresh

count = st_autorefresh(interval=5000, key="live_refresh")
"""

# Python: FastAPI with JWT Authentication
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel
from datetime import datetime, timedelta
from typing import Optional
import os

# Configuration
SECRET_KEY = os.getenv("SECRET_KEY", "your-secret-key-change-in-production")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

app = FastAPI()

# Password hashing
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# Models
class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: Optional[str] = None
    role: Optional[str] = None

class User(BaseModel):
    username: str
    email: str
    role: str
    disabled: bool = False

class UserInDB(User):
    hashed_password: str

# Simulated user database (use real DB in production)
fake_users_db = {
    "analyst": {
        "username": "analyst",
        "email": "analyst@club.com",
        "role": "analyst",
        "hashed_password": pwd_context.hash("analyst123"),
        "disabled": False,
    },
    "manager": {
        "username": "manager",
        "email": "manager@club.com",
        "role": "manager",
        "hashed_password": pwd_context.hash("manager123"),
        "disabled": False,
    },
    "admin": {
        "username": "admin",
        "email": "admin@club.com",
        "role": "admin",
        "hashed_password": pwd_context.hash("admin123"),
        "disabled": False,
    }
}

# Helper functions
def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def get_user(db: dict, username: str) -> Optional[UserInDB]:
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)
    return None

def authenticate_user(db: dict, username: str, password: str) -> Optional[UserInDB]:
    user = get_user(db, username)
    if not user or not verify_password(password, user.hashed_password):
        return None
    return user

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        role: str = payload.get("role")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username, role=role)
    except JWTError:
        raise credentials_exception

    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

# Role-based access decorator
def require_role(allowed_roles: list):
    async def role_checker(current_user: User = Depends(get_current_user)):
        if current_user.role not in allowed_roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Insufficient permissions"
            )
        return current_user
    return role_checker

# Endpoints
@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(
        data={"sub": user.username, "role": user.role},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

# Protected endpoints with role requirements
@app.get("/admin/users")
async def admin_get_users(user: User = Depends(require_role(["admin"]))):
    """Admin only - get all users"""
    return {"users": list(fake_users_db.keys())}

@app.get("/manager/team/{team_id}")
async def manager_get_team(
    team_id: int,
    user: User = Depends(require_role(["admin", "manager"]))
):
    """Managers and admins - get team details"""
    return {"team_id": team_id, "accessed_by": user.username}

@app.get("/analytics/players")
async def get_players(user: User = Depends(get_current_user)):
    """All authenticated users - get player analytics"""
    return {"players": ["Player 1", "Player 2"], "user_role": user.role}

# Python: Performance Optimization Techniques
import streamlit as st
import pandas as pd
import numpy as np
from functools import lru_cache
import asyncio
from concurrent.futures import ThreadPoolExecutor
import dask.dataframe as dd

# 1. Streamlit caching
@st.cache_data(ttl=3600)  # Cache for 1 hour
def load_large_dataset(filepath: str) -> pd.DataFrame:
    """Load and cache large dataset"""
    return pd.read_parquet(filepath)

@st.cache_resource  # Cache resource objects (DB connections, models)
def get_database_connection():
    """Cache database connection"""
    return create_engine("postgresql://...")

# 2. Efficient data loading with chunking
def process_large_csv(filepath: str, chunk_size: int = 100000):
    """Process large CSV in chunks"""
    results = []

    for chunk in pd.read_csv(filepath, chunksize=chunk_size):
        # Process each chunk
        chunk_result = chunk.groupby("team").agg({
            "goals": "sum",
            "xG": "mean"
        })
        results.append(chunk_result)

    # Combine results
    return pd.concat(results).groupby(level=0).sum()

# 3. Dask for out-of-core computation
def analyze_with_dask(filepath: str):
    """Use Dask for large-than-memory datasets"""
    # Dask reads data lazily
    ddf = dd.read_parquet(filepath)

    # Operations are lazy until .compute()
    result = ddf.groupby("team").agg({
        "goals": "sum",
        "xG": "mean"
    }).compute()

    return result

# 4. Async operations for I/O bound tasks
async def fetch_multiple_apis(endpoints: list) -> list:
    """Fetch from multiple APIs concurrently"""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_data(session, url) for url in endpoints]
        return await asyncio.gather(*tasks)

async def fetch_data(session, url):
    async with session.get(url) as response:
        return await response.json()

# 5. ThreadPoolExecutor for CPU-bound tasks
def parallel_processing(data_chunks: list):
    """Process data chunks in parallel"""
    with ThreadPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(process_chunk, data_chunks))
    return pd.concat(results)

def process_chunk(chunk):
    # CPU-intensive processing
    return chunk.apply(complex_calculation, axis=1)

# 6. Query optimization with SQL pushdown
def optimized_query(engine, team_id: int, season: str):
    """Push filtering to database"""
    # Bad: Load all data then filter
    # df = pd.read_sql("SELECT * FROM matches", engine)
    # df = df[(df.team_id == team_id) & (df.season == season)]

    # Good: Filter in SQL
    query = """
        SELECT m.*, p.name as player_name
        FROM matches m
        JOIN players p ON m.player_id = p.id
        WHERE m.team_id = %(team_id)s
          AND m.season = %(season)s
    """
    return pd.read_sql(query, engine, params={"team_id": team_id, "season": season})

# 7. Memory optimization
def optimize_memory(df: pd.DataFrame) -> pd.DataFrame:
    """Reduce DataFrame memory usage"""

    for col in df.columns:
        col_type = df[col].dtype

        if col_type == "object":
            # Convert to category if low cardinality
            if df[col].nunique() / len(df) < 0.5:
                df[col] = df[col].astype("category")

        elif col_type == "int64":
            # Downcast integers
            df[col] = pd.to_numeric(df[col], downcast="integer")

        elif col_type == "float64":
            # Downcast floats
            df[col] = pd.to_numeric(df[col], downcast="float")

    return df

# Usage in Streamlit
# Before: df = pd.read_csv("large_file.csv")
# After:
# df = load_large_dataset("data.parquet")
# df = optimize_memory(df)

# Python: Testing Analytics Applications
import pytest
from fastapi.testclient import TestClient
from unittest.mock import Mock, patch
import pandas as pd
import numpy as np

# Import your app
from main import app
from data_layer import DataManager
from analytics import calculate_xG, PlayerAnalyzer

client = TestClient(app)

# 1. Unit tests for analytics functions
class TestXGCalculation:
    def test_basic_xg_calculation(self):
        shots = pd.DataFrame({
            "x": [0.9, 0.85, 0.7],
            "y": [0.5, 0.3, 0.6],
            "shot_type": ["foot", "head", "foot"]
        })

        result = calculate_xG(shots)

        assert len(result) == 3
        assert all(0 <= xg <= 1 for xg in result)
        assert result[0] > result[2]  # Closer shot higher xG

    def test_edge_cases(self):
        # Empty dataframe
        empty_shots = pd.DataFrame(columns=["x", "y", "shot_type"])
        result = calculate_xG(empty_shots)
        assert len(result) == 0

        # Single shot
        single = pd.DataFrame({"x": [0.88], "y": [0.5], "shot_type": ["foot"]})
        result = calculate_xG(single)
        assert len(result) == 1

class TestPlayerAnalyzer:
    @pytest.fixture
    def sample_data(self):
        return pd.DataFrame({
            "player": ["A", "B", "C"],
            "position": ["FW", "MF", "DF"],
            "goals": [10, 5, 2],
            "xG": [8.5, 6.0, 1.5]
        })

    def test_filter_by_position(self, sample_data):
        analyzer = PlayerAnalyzer(sample_data)
        forwards = analyzer.filter_by_position("FW")

        assert len(forwards) == 1
        assert forwards.iloc[0]["player"] == "A"

    def test_goal_overperformance(self, sample_data):
        analyzer = PlayerAnalyzer(sample_data)
        result = analyzer.calculate_overperformance()

        assert result.loc[result["player"] == "A", "overperformance"].values[0] == 1.5

# 2. API testing with FastAPI TestClient
class TestPlayersAPI:
    def test_get_all_players(self):
        response = client.get("/players")
        assert response.status_code == 200
        assert isinstance(response.json(), list)

    def test_get_player_by_id(self):
        response = client.get("/players/1")
        assert response.status_code == 200
        assert "name" in response.json()

    def test_player_not_found(self):
        response = client.get("/players/9999")
        assert response.status_code == 404

    def test_filter_players(self):
        response = client.get("/players?position=FW&min_goals=5")
        assert response.status_code == 200
        players = response.json()
        assert all(p["position"] == "FW" for p in players)
        assert all(p["goals"] >= 5 for p in players)

class TestAuthentication:
    def test_login_success(self):
        response = client.post("/token",
            data={"username": "analyst", "password": "analyst123"})
        assert response.status_code == 200
        assert "access_token" in response.json()

    def test_login_failure(self):
        response = client.post("/token",
            data={"username": "analyst", "password": "wrong"})
        assert response.status_code == 401

    def test_protected_endpoint_without_token(self):
        response = client.get("/admin/users")
        assert response.status_code == 401

    def test_protected_endpoint_with_token(self):
        # First login
        login = client.post("/token",
            data={"username": "admin", "password": "admin123"})
        token = login.json()["access_token"]

        # Then access protected endpoint
        response = client.get("/admin/users",
            headers={"Authorization": f"Bearer {token}"})
        assert response.status_code == 200

# 3. Integration tests with mocking
class TestDataIntegration:
    @patch("data_layer.DataManager.get_players")
    def test_player_endpoint_with_mock(self, mock_get_players):
        mock_get_players.return_value = pd.DataFrame({
            "id": [1, 2],
            "name": ["Player 1", "Player 2"],
            "goals": [10, 5]
        })

        response = client.get("/players")

        assert response.status_code == 200
        assert len(response.json()) == 2
        mock_get_players.assert_called_once()

# 4. Streamlit testing with pytest
def test_streamlit_components():
    """Test Streamlit helper functions"""
    from app import create_radar_chart, format_player_stats

    # Test chart creation
    player_data = {"shooting": 80, "passing": 75, "defending": 60}
    chart = create_radar_chart(player_data)
    assert chart is not None

    # Test formatting
    stats = format_player_stats({"goals": 10, "assists": 5})
    assert "Goals: 10" in stats

# Run with: pytest tests/ -v --cov=app

# Python: Complete FastAPI + Streamlit Template
"""
Project structure:
/app
  /api
    main.py       # FastAPI app
    auth.py       # Authentication
    models.py     # Pydantic models
  /dashboard
    app.py        # Streamlit app
  /data
    manager.py    # Data layer
  /tests
    test_api.py
  docker-compose.yml
  requirements.txt
"""

# api/main.py
from fastapi import FastAPI, Depends, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from .auth import get_current_user
from .models import Player, Team
from ..data.manager import DataManager

app = FastAPI(
    title="Football Analytics API",
    version="1.0.0",
    docs_url="/docs"
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Configure for production
    allow_methods=["*"],
    allow_headers=["*"],
)

dm = DataManager()

@app.get("/health")
async def health_check():
    return {"status": "healthy", "version": "1.0.0"}

@app.get("/players", response_model=list[Player])
async def get_players(
    season: str = "2023-24",
    user = Depends(get_current_user)
):
    return dm.get_players(season)

@app.get("/teams/{team_id}")
async def get_team(team_id: int, user = Depends(get_current_user)):
    team = dm.get_team(team_id)
    if not team:
        raise HTTPException(404, "Team not found")
    return team

# dashboard/app.py
import streamlit as st
import requests
import plotly.express as px

st.set_page_config(page_title="Football Analytics", layout="wide")

# Authentication
if "token" not in st.session_state:
    with st.form("login"):
        username = st.text_input("Username")
        password = st.text_input("Password", type="password")
        if st.form_submit_button("Login"):
            response = requests.post(
                "http://api:8000/token",
                data={"username": username, "password": password}
            )
            if response.ok:
                st.session_state.token = response.json()["access_token"]
                st.rerun()
            else:
                st.error("Invalid credentials")
    st.stop()

# Headers for API requests
headers = {"Authorization": f"Bearer {st.session_state.token}"}

# Sidebar
season = st.sidebar.selectbox("Season", ["2023-24", "2022-23"])

# Main content
st.title("Football Analytics Dashboard")

# Fetch data
players = requests.get(
    f"http://api:8000/players?season={season}",
    headers=headers
).json()

# KPIs
col1, col2, col3 = st.columns(3)
col1.metric("Players", len(players))
col2.metric("Total Goals", sum(p["goals"] for p in players))
col3.metric("Avg xG", f"{sum(p['xG'] for p in players)/len(players):.2f}")

# Chart
fig = px.scatter(players, x="xG", y="goals", hover_name="name")
st.plotly_chart(fig, use_container_width=True)

# data/manager.py
from functools import lru_cache
import pandas as pd
from sqlalchemy import create_engine
import redis
import json
import os

class DataManager:
    def __init__(self):
        self.engine = create_engine(os.getenv("DATABASE_URL"))
        self.redis = redis.from_url(os.getenv("REDIS_URL"))

    def get_players(self, season: str) -> list:
        cache_key = f"players:{season}"

        # Check cache
        cached = self.redis.get(cache_key)
        if cached:
            return json.loads(cached)

        # Query database
        df = pd.read_sql(
            "SELECT * FROM players WHERE season = %s",
            self.engine,
            params=[season]
        )
        result = df.to_dict("records")

        # Cache result
        self.redis.setex(cache_key, 3600, json.dumps(result))

        return result