Capstone - Complete Analytics System

import pandas as pd
from tabulate import tabulate

# Football Analytics Research Taxonomy
research_areas = pd.DataFrame({
    "area": ["Expected Goals Models", "Player Valuation", "Tactical Analysis",
             "Tracking Data", "Injury Prediction", "Match Outcome Prediction",
             "Player Similarity", "Team Style Clustering", "Set Piece Analysis",
             "Goalkeeper Analysis"],
    "maturity": ["High", "High", "Medium-High", "Medium", "Medium",
                 "High", "Medium-High", "Medium", "Medium", "Medium"],
    "key_methods": ["Logistic Regression, XGBoost, Neural Nets",
                    "Market Models, Performance Metrics",
                    "Network Analysis, Clustering",
                    "Computer Vision, Spatial Statistics",
                    "Survival Analysis, Time Series",
                    "Poisson Models, Machine Learning",
                    "Embedding, Distance Metrics",
                    "K-Means, Hierarchical Clustering",
                    "Expected Goals, Game Theory",
                    "GSAA, Positioning Models"],
    "data_requirements": ["Event Data", "Event + Market Data", "Event/Tracking Data",
                          "Tracking Data", "Medical + Physical Data",
                          "Historical Match Data", "Event Data",
                          "Event/Tracking Data", "Event Data", "Event/Tracking Data"],
    "industry_adoption": ["Universal", "High", "Growing", "Elite Clubs",
                          "Growing", "Betting Industry", "Recruitment",
                          "Analysis Teams", "Set Piece Coaches", "Growing"]
})

print("Football Analytics Research Areas")
print("=" * 100)
print(tabulate(research_areas, headers="keys", tablefmt="grid", showindex=False))

# Research opportunity scoring
print("\n\nResearch Opportunity Assessment:")
maturity_scores = {"High": 1, "Medium-High": 2, "Medium": 3, "Low": 4}
research_areas["opportunity_score"] = research_areas["maturity"].map(maturity_scores)

opportunities = research_areas.nlargest(5, "opportunity_score")[["area", "maturity", "key_methods"]]
print("\nHigh-Opportunity Research Areas (Less Mature = More Opportunity):")
print(opportunities.to_string(index=False))

import pandas as pd
import numpy as np
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime

# Research Project Framework

@dataclass
class ResearchProject:
    """Framework for conducting rigorous football analytics research."""

    title: str
    hypothesis: str
    status: str = "planning"
    data_sources: Dict = field(default_factory=dict)
    methodology: Dict = field(default_factory=dict)
    results: Dict = field(default_factory=dict)

    def __post_init__(self):
        print(f"Research project initialized: {self.title}")

    def conduct_literature_review(self, keywords: List[str],
                                   databases: List[str] = None) -> Dict:
        """Phase 1: Systematic literature review."""
        print("Conducting literature review...")

        if databases is None:
            databases = ["Google Scholar", "arXiv", "SSRN"]

        search_strategy = {
            "keywords": keywords,
            "databases": databases,
            "date_range": (2015, 2024),
            "inclusion_criteria": [
                "Peer-reviewed or reputable preprint",
                "Football/soccer specific or methodology transferable",
                "Publicly available or accessible"
            ],
            "exclusion_criteria": [
                "Non-English publications",
                "Conference abstracts only",
                "Superseded by newer work"
            ]
        }

        self.methodology["literature_review"] = search_strategy
        print("Literature review strategy documented")

        return search_strategy

    def setup_data_pipeline(self, sources: List[str],
                            validation_checks: List[str]) -> Dict:
        """Phase 2: Data collection and validation setup."""
        print("Setting up data pipeline...")

        data_pipeline = {
            "sources": sources,
            "collection_date": datetime.now().isoformat(),
            "validation": validation_checks,
            "preprocessing_steps": [
                "Remove duplicates",
                "Handle missing values",
                "Validate data ranges",
                "Cross-reference with external sources"
            ]
        }

        self.data_sources = data_pipeline
        self.status = "data_collection"

        return data_pipeline

    def design_methodology(self, approach: str,
                           baseline_comparisons: List[str],
                           metrics: List[str]) -> Dict:
        """Phase 3: Methodology design."""
        print("Designing methodology...")

        methodology = {
            "approach": approach,
            "baselines": baseline_comparisons,
            "evaluation_metrics": metrics,
            "statistical_tests": ["t-test", "Mann-Whitney U", "Bootstrap CI"],
            "cross_validation": {
                "method": "5-fold stratified",
                "holdout_set": 0.2,
                "temporal_split": True
            }
        }

        self.methodology["analysis"] = methodology
        self.status = "methodology"

        return methodology

    def run_experiments(self, model_configs: List[str]) -> pd.DataFrame:
        """Phase 4: Run experiments."""
        print("Running experiments...")

        experiments = pd.DataFrame({
            "experiment_id": [f"exp_{i+1}" for i in range(len(model_configs))],
            "config": model_configs,
            "status": "pending",
            "start_time": None,
            "end_time": None,
            "primary_metric": None
        })

        self.results["experiments"] = experiments
        self.status = "experimentation"

        return experiments

    def validate_results(self, results_df: pd.DataFrame,
                         alpha: float = 0.05) -> Dict:
        """Phase 5: Statistical validation."""
        print("Validating results...")

        validation = {
            "sample_size": len(results_df),
            "statistical_power": None,
            "confidence_level": 1 - alpha,
            "tests_performed": [],
            "multiple_comparison_correction": "Bonferroni",
            "effect_sizes": {}
        }

        self.results["validation"] = validation
        self.status = "validation"

        return validation

    def generate_summary(self) -> Dict:
        """Generate research summary."""
        return {
            "title": self.title,
            "hypothesis": self.hypothesis,
            "status": self.status,
            "data_sources": len(self.data_sources.get("sources", [])),
            "methodology": self.methodology.get("analysis", {}).get("approach"),
            "key_findings": self.results
        }


# Example: Setting up an xG research project
xg_project = ResearchProject(
    title="Contextual Expected Goals: Incorporating Defensive Pressure",
    hypothesis="Adding defensive pressure features improves xG model accuracy by >5%"
)

# Literature review
lit_review = xg_project.conduct_literature_review(
    keywords=["expected goals", "xG", "defensive pressure", "football analytics"],
    databases=["Google Scholar", "arXiv", "MIT Sloan Archives"]
)

# Data setup
data_pipeline = xg_project.setup_data_pipeline(
    sources=["StatsBomb Open Data", "Wyscout", "Custom Tracking"],
    validation_checks=["Shot count verification", "xG range validation",
                       "Missing coordinate check"]
)

# Methodology
methodology = xg_project.design_methodology(
    approach="Gradient Boosting with feature engineering",
    baseline_comparisons=["Basic xG (distance/angle)", "StatsBomb xG", "Public models"],
    metrics=["Log Loss", "Brier Score", "AUC-ROC", "Calibration"]
)

print("\nResearch Project Summary:")
print("=" * 50)
summary = xg_project.generate_summary()
for key, value in summary.items():
    print(f"{key}: {value}")

import pandas as pd
import numpy as np
from scipy import stats

# Creating Publication-Quality Tables and Statistics

def create_results_table(results_data: pd.DataFrame) -> str:
    """
    Create a formatted results table for publication.
    """

    # Highlight best performance
    best_idx = results_data["log_loss"].idxmin()

    # Format table
    table_str = """
    ╔════════════════════════════════════════════════════════════════════════╗
    ║     Expected Goals Model Performance Comparison                        ║
    ╠═══════════════════════╦═══════════╦═════════════╦═════════╦════════════╣
    ║ Model                 ║ Log Loss  ║ Brier Score ║ AUC-ROC ║ Calibration║
    ╠═══════════════════════╬═══════════╬═════════════╬═════════╬════════════╣"""

    for idx, row in results_data.iterrows():
        marker = " *" if idx == best_idx else "  "
        table_str += f"""
    ║ {row['model']:<21}║ {row['log_loss']:.4f}{marker} ║ {row['brier_score']:.4f}     ║ {row['auc_roc']:.3f}  ║ {row['calibration']:.4f}    ║"""

    table_str += """
    ╚═══════════════════════╩═══════════╩═════════════╩═════════╩════════════╝
    * Best performance (all differences significant at p < 0.05)
    """

    return table_str


# Example results data
model_results = pd.DataFrame({
    "model": ["Baseline (Dist/Angle)", "Random Forest", "XGBoost",
              "Neural Network", "Our Model (Pressure)"],
    "log_loss": [0.3421, 0.3156, 0.3089, 0.3102, 0.2934],
    "brier_score": [0.0923, 0.0867, 0.0842, 0.0851, 0.0798],
    "auc_roc": [0.762, 0.789, 0.801, 0.798, 0.824],
    "calibration": [0.0312, 0.0245, 0.0198, 0.0221, 0.0156]
})

print(create_results_table(model_results))


def report_significance(metric_a: np.ndarray, metric_b: np.ndarray,
                        test_type: str = "paired_t") -> str:
    """
    Generate statistical significance report.
    """

    # Perform statistical test
    if test_type == "paired_t":
        stat, p_value = stats.ttest_rel(metric_a, metric_b)
    else:  # wilcoxon
        stat, p_value = stats.wilcoxon(metric_a, metric_b)

    # Effect size (Cohen's d)
    diff = metric_a - metric_b
    cohens_d = np.mean(diff) / np.std(diff)

    # Bootstrap confidence interval
    n_bootstrap = 1000
    improvements = []
    for _ in range(n_bootstrap):
        idx = np.random.choice(len(metric_a), len(metric_a), replace=True)
        improvements.append(
            (1 - np.mean(metric_b[idx]) / np.mean(metric_a[idx])) * 100
        )

    ci_low, ci_high = np.percentile(improvements, [2.5, 97.5])

    # Format p-value
    p_str = "< 0.001" if p_value < 0.001 else f"= {p_value:.3f}"

    return (f"Improvement: {(1 - np.mean(metric_b) / np.mean(metric_a)) * 100:.1f}% "
            f"(95% CI: [{ci_low:.1f}%, {ci_high:.1f}%], p {p_str}, d = {cohens_d:.2f})")


# Example comparison
np.random.seed(42)
baseline_scores = np.random.normal(0.34, 0.05, 100)
improved_scores = np.random.normal(0.29, 0.05, 100)

print("\nStatistical Report:")
print(report_significance(baseline_scores, improved_scores))

import pandas as pd
import numpy as np
import hashlib
import json
from typing import Dict, Any
import sys

# Reproducibility Best Practices

def setup_reproducible_environment() -> Dict[str, Any]:
    """
    Document environment for reproducibility.
    """

    environment_doc = {
        "python_version": sys.version,
        "platform": sys.platform,
        "packages": {},  # Would use pip freeze in practice
        "snapshot_date": pd.Timestamp.now().isoformat()
    }

    # In practice, use:
    # pip freeze > requirements.txt
    # or
    # conda env export > environment.yml

    return environment_doc


def create_data_dictionary(dataset: pd.DataFrame,
                           dataset_name: str) -> pd.DataFrame:
    """
    Generate comprehensive data dictionary.
    """

    dictionary = pd.DataFrame({
        "variable": dataset.columns,
        "type": [str(dtype) for dtype in dataset.dtypes],
        "n_unique": [dataset[col].nunique() for col in dataset.columns],
        "n_missing": [dataset[col].isna().sum() for col in dataset.columns],
        "pct_missing": [dataset[col].isna().mean() * 100 for col in dataset.columns],
        "example_values": [
            ", ".join(map(str, dataset[col].dropna().unique()[:3]))
            for col in dataset.columns
        ]
    })

    # Save to file
    # dictionary.to_csv(f"{dataset_name}_dictionary.csv", index=False)

    return dictionary


def document_analysis_pipeline() -> Dict[str, Any]:
    """
    Document the analysis pipeline for reproducibility.
    """

    pipeline = {
        "steps": [
            "1. Data Loading: src/load_raw_data.py",
            "2. Preprocessing: src/preprocess_events.py",
            "3. Feature Engineering: src/create_features.py",
            "4. Model Training: src/train_model.py",
            "5. Evaluation: src/evaluate_model.py",
            "6. Visualization: src/create_figures.py"
        ],
        "execution_order": "Run python main.py or use make",
        "expected_runtime": "~2 hours on standard hardware",
        "hardware_requirements": "16GB RAM recommended for full dataset"
    }

    return pipeline


def create_results_checksum(results_df: pd.DataFrame) -> Dict[str, Any]:
    """
    Create checksum for results verification.
    """

    # Select numeric columns
    numeric_cols = results_df.select_dtypes(include=[np.number])

    checksum = {
        "n_rows": len(results_df),
        "n_cols": len(results_df.columns),
        "column_sums": numeric_cols.sum().to_dict(),
        "md5_hash": hashlib.md5(
            pd.util.hash_pandas_object(results_df).values
        ).hexdigest()
    }

    return checksum


# Example usage
np.random.seed(42)
example_data = pd.DataFrame({
    "shot_id": range(1, 1001),
    "xg": np.random.uniform(0, 1, 1000),
    "goal": np.random.binomial(1, 0.1, 1000),
    "distance": np.random.uniform(5, 35, 1000)
})

# Create documentation
print("Data Dictionary:")
print("=" * 70)
data_dict = create_data_dictionary(example_data, "shots")
print(data_dict.to_string(index=False))

print("\n\nAnalysis Pipeline:")
print("=" * 70)
pipeline_doc = document_analysis_pipeline()
for key, value in pipeline_doc.items():
    print(f"\n{key}:")
    if isinstance(value, list):
        for item in value:
            print(f"  {item}")
    else:
        print(f"  {value}")

print("\n\nResults Checksum:")
print("=" * 70)
checksum = create_results_checksum(example_data)
print(json.dumps(checksum, indent=2))

import pandas as pd
from dataclasses import dataclass
from typing import List, Dict

# Industry Publication Framework

@dataclass
class IndustryPublication:
    """Framework for creating industry-focused content."""

    title: str
    format: str
    audience: str

    def create_blog_outline(self, key_finding: str,
                            data_viz_count: int = 3) -> Dict:
        """Create blog post outline."""

        outline = {
            "hook": "Compelling opening that challenges conventional wisdom",
            "context": "Brief background (2-3 paragraphs)",
            "methodology": "High-level approach without excessive technical detail",
            "key_findings": {
                "finding_1": "Primary insight with supporting visualization",
                "finding_2": "Secondary insight",
                "finding_3": "Practical implication"
            },
            "visualizations": f"{data_viz_count} clear, annotated charts",
            "conclusion": "Actionable takeaway for readers",
            "technical_appendix": "Optional: detailed methodology"
        }

        return outline

    def create_talk_outline(self, duration_minutes: int = 20) -> Dict:
        """Create conference talk outline."""

        outline = {
            "intro": {
                "duration": "2 minutes",
                "content": ["Hook/problem statement", "Why this matters",
                           "Preview of key insight"]
            },
            "context": {
                "duration": "3 minutes",
                "content": ["Industry background", "Current approaches",
                           "Gap/opportunity"]
            },
            "methodology": {
                "duration": "5 minutes",
                "content": ["Data sources", "Analytical approach",
                           "Key innovations"]
            },
            "results": {
                "duration": "7 minutes",
                "content": ["Main findings (3 max)", "Visualizations",
                           "Statistical support"]
            },
            "implications": {
                "duration": "2 minutes",
                "content": ["Practical applications", "Limitations",
                           "Future directions"]
            },
            "qa": {
                "duration": "1 minute buffer",
                "content": "Prepare for common questions"
            }
        }

        return outline

    def create_twitter_thread(self, key_points: List[str],
                              max_tweets: int = 10) -> Dict:
        """Create Twitter/X thread outline."""

        thread = {
            "tweet_1": "Hook: Surprising finding or question",
            "tweets_middle": [f"Key point {i+1}" for i in range(min(len(key_points), max_tweets - 2))],
            "final_tweet": "Call to action: Link to full analysis",
            "media": "Include 1-2 compelling visualizations",
            "hashtags": ["#FootballAnalytics", "#xG", "#SportScience"]
        }

        return thread


# Example: Creating content for different formats
content = IndustryPublication(
    title="Why Traditional Assists Are Misleading",
    format="multi-platform",
    audience="Football analysts and enthusiasts"
)

blog_outline = content.create_blog_outline(
    key_finding="xA provides 40% better prediction of future assists"
)

talk_outline = content.create_talk_outline(duration_minutes=15)

print("Blog Post Outline")
print("=" * 50)
for section, description in blog_outline.items():
    print(f"\n{section}:")
    if isinstance(description, dict):
        for k, v in description.items():
            print(f"  - {k}: {v}")
    else:
        print(f"  {description}")

print("\n\nConference Talk Outline")
print("=" * 50)
for section, details in talk_outline.items():
    print(f"\n{section} ({details['duration']}):")
    if isinstance(details["content"], list):
        for item in details["content"]:
            print(f"  - {item}")
    else:
        print(f"  {details['content']}")

import pandas as pd
from dataclasses import dataclass, field
from typing import List, Dict, Optional

# Research Portfolio Tracker

@dataclass
class ResearchPortfolio:
    """Track research outputs and impact."""

    publications: pd.DataFrame = field(default_factory=lambda: pd.DataFrame(
        columns=["title", "venue", "year", "type", "citations", "url"]
    ))
    projects: pd.DataFrame = field(default_factory=lambda: pd.DataFrame(
        columns=["name", "description", "github_stars", "downloads", "status"]
    ))
    talks: pd.DataFrame = field(default_factory=lambda: pd.DataFrame(
        columns=["title", "venue", "date", "audience_size", "recording_url"]
    ))
    collaborations: pd.DataFrame = field(default_factory=lambda: pd.DataFrame(
        columns=["partner", "organization", "project", "status"]
    ))
    metrics: Dict = field(default_factory=lambda: {
        "h_index": None,
        "total_citations": None,
        "github_followers": None,
        "twitter_followers": None
    })

    def add_publication(self, title: str, venue: str, year: int,
                        pub_type: str, citations: int = 0,
                        url: str = "") -> None:
        """Add a publication to portfolio."""
        new_pub = pd.DataFrame([{
            "title": title, "venue": venue, "year": year,
            "type": pub_type, "citations": citations, "url": url
        }])
        self.publications = pd.concat([self.publications, new_pub],
                                       ignore_index=True)

    def add_project(self, name: str, description: str,
                    github_stars: int = 0, downloads: int = 0,
                    status: str = "active") -> None:
        """Add an open-source project."""
        new_project = pd.DataFrame([{
            "name": name, "description": description,
            "github_stars": github_stars, "downloads": downloads,
            "status": status
        }])
        self.projects = pd.concat([self.projects, new_project],
                                   ignore_index=True)

    def calculate_impact(self) -> Dict:
        """Calculate impact metrics."""
        return {
            "academic_impact": {
                "publications": len(self.publications),
                "peer_reviewed": len(self.publications[
                    self.publications["type"] == "peer-reviewed"
                ]),
                "total_citations": self.publications["citations"].sum(),
                "h_index": self.metrics.get("h_index")
            },
            "industry_impact": {
                "blog_posts": len(self.publications[
                    self.publications["type"] == "blog"
                ]),
                "talks_given": len(self.talks),
                "open_source_projects": len(self.projects),
                "total_github_stars": self.projects["github_stars"].sum()
            },
            "community_impact": {
                "collaborations": len(self.collaborations),
                "mentees": None,
                "datasets_released": None
            }
        }


# Example portfolio
my_portfolio = ResearchPortfolio()

# Add a publication
my_portfolio.add_publication(
    title="Contextual Expected Goals with Defensive Pressure",
    venue="MIT Sloan Sports Analytics Conference",
    year=2024,
    pub_type="peer-reviewed",
    citations=15,
    url="https://example.com/paper"
)

# Add an open-source project
my_portfolio.add_project(
    name="football-xg",
    description="Open-source expected goals model library",
    github_stars=245,
    downloads=12000,
    status="active"
)

# Calculate impact
impact = my_portfolio.calculate_impact()

print("Research Impact Summary")
print("=" * 50)
for category, metrics in impact.items():
    print(f"\n{category}:")
    for metric, value in metrics.items():
        print(f"  {metric}: {value}")