#!/usr/bin/env python3 """ Trifecta Pro — Hugging Face Hub Integration Script 1. Uploads trained model artifact (trifecta_pro_optimized_model.joblib) to HF Model Hub. 2. Uploads training dataset (data/fair_dinkum_training_dataset.csv) to HF Dataset Hub. 3. Scaffolds and pushes a live Gradio prediction app with Auto-Fetch All Australian Gallops Meetings. """ import os import sys from pathlib import Path from huggingface_hub import HfApi, create_repo def get_hf_token(): token = os.environ.get("HF_TOKEN") or os.environ.get("HUGGINGFACE_TOKEN") return token def push_model_to_hub(repo_id: str = "brettapps1/trifecta-pro-model"): print(f"📦 Uploading model artifact to Hugging Face Model Hub: {repo_id}") token = get_hf_token() api = HfApi(token=token) try: create_repo(repo_id=repo_id, repo_type="model", private=False, exist_ok=True, token=token) except Exception as e: print(f"Note: {e}") model_path = Path("trifecta_pro_optimized_model.joblib") if model_path.exists(): api.upload_file( path_or_fileobj=str(model_path), path_in_repo="trifecta_pro_optimized_model.joblib", repo_id=repo_id, repo_type="model", token=token ) print(f"✅ Model uploaded to https://huggingface.co/{repo_id}") def push_dataset_to_hub(repo_id: str = "brettapps1/trifecta-pro-dataset"): print(f"📊 Uploading training dataset to Hugging Face Dataset Hub: {repo_id}") token = get_hf_token() api = HfApi(token=token) try: create_repo(repo_id=repo_id, repo_type="dataset", private=False, exist_ok=True, token=token) except Exception as e: print(f"Note: {e}") dataset_path = Path("data/fair_dinkum_training_dataset.csv") if dataset_path.exists(): api.upload_file( path_or_fileobj=str(dataset_path), path_in_repo="fair_dinkum_training_dataset.csv", repo_id=repo_id, repo_type="dataset", token=token ) print(f"✅ Dataset uploaded to https://huggingface.co/datasets/{repo_id}") def deploy_hf_space(space_id: str = "Brettapps/trifecta-pro-space"): print(f"🚀 Deploying Gradio Trifecta App with Auto-Fetch All Meetings to HF Space: {space_id}") token = get_hf_token() api = HfApi(token=token) try: create_repo(repo_id=space_id, repo_type="space", space_sdk="gradio", private=False, exist_ok=True, token=token) except Exception as e: print(f"Note: {e}") space_dir = Path("outputs/hf_space_temp") space_dir.mkdir(parents=True, exist_ok=True) readme_content = """--- title: Trifecta Pro - FormFav Racing Intelligence emoji: 🐎 colorFrom: blue colorTo: indigo sdk: gradio sdk_version: 4.44.0 app_file: app.py pinned: false --- # Trifecta Pro Quant Racing Intelligence - Auto-Fetch Australian Gallops """ (space_dir / "README.md").write_text(readme_content) reqs_content = """scikit-learn>=1.3.0 numpy>=1.24.0 gradio>=4.0.0 joblib>=1.3.0 pandas>=2.0.0 httpx>=0.24.0 """ (space_dir / "requirements.txt").write_text(reqs_content) app_py_content = """import gradio as gr import numpy as np import joblib import os import httpx MODEL_PATH = "trifecta_pro_optimized_model.joblib" def load_model(): if os.path.exists(MODEL_PATH): return joblib.load(MODEL_PATH) return None model = load_model() def fetch_all_aus_meetings_and_predict(api_key: str, date_str: str): if not api_key: return "❌ Error: FormFav API Key required. Get one at https://formfav.com" headers = {"X-API-Key": api_key.strip()} meetings_url = f"https://api.formfav.com/v1/meetings/{date_str.strip()}" try: with httpx.Client(timeout=25.0) as client: resp = client.get(meetings_url, headers=headers) if resp.status_code != 200: return f"❌ FormFav API Error (Status {resp.status_code}): {resp.text}" data = resp.json() except Exception as e: return f"❌ Connection Error: {str(e)}" meetings = data.get("meetings", []) au_meetings = [m for m in meetings if m.get("country", "").lower() in ["au", "australia"]] if not au_meetings: return f"⚠️ No Australian gallops meetings found for {date_str}." if model is None: return "❌ Model artifact not loaded." report = [] report.append(f"# 🎯 Australian Gallops Trifecta Pro Report — {date_str}\\n") report.append(f"Found {len(au_meetings)} Australian meetings. Running ML predictions across all races...\\n") for m in au_meetings: track_name = m.get("track", m.get("slug", "Unknown Track")) track_slug = m.get("slug", "") report.append(f"## 🏟️ Meeting: {track_name} ({track_slug.upper()})\\n") races = m.get("races", []) if not races: report.append("- *Races data pending or unavailable.*\\n") continue for r in races: rnum = r.get("raceNumber", 1) rname = r.get("raceName", f"Race {rnum}") dist = r.get("distance", "N/A") cond = r.get("condition", "N/A") report.append(f"### Race {rnum}: {rname} ({dist} | {cond})") runners = r.get("runners", []) if not runners and track_slug: card_url = f"https://api.formfav.com/v1/race-card/{date_str}/{track_slug}/{rnum}" try: with httpx.Client(timeout=10.0) as client: c_resp = client.get(card_url, headers=headers) if c_resp.status_code == 200: runners = c_resp.json().get("runners", []) except Exception: pass if runners: runners_features = [] runner_names = [] for runner in runners: name = runner.get("horseName", "Unknown") runner_names.append(name) barrier = int(runner.get("barrier", 8)) if str(runner.get("barrier", "")).isdigit() else 8 weight = float(runner.get("weight", 56.0)) if str(runner.get("weight", "")).replace(".","").isdigit() else 56.0 rating = int(runner.get("rating", 75)) if str(runner.get("rating", "")).isdigit() else 75 recent_finish = float(runner.get("recentFinish", 3.0)) if str(runner.get("recentFinish", "")).replace(".","").isdigit() else 3.0 track_match = 1.0 if runner.get("trackMatch", True) else 0.0 dist_sr = float(runner.get("distanceStrikeRate", 0.3)) if str(runner.get("distanceStrikeRate", "")).replace(".","").isdigit() else 0.3 trainer_form = float(runner.get("trainerForm", 0.2)) if str(runner.get("trainerForm", "")).replace(".","").isdigit() else 0.2 jockey_form = float(runner.get("jockeyForm", 0.2)) if str(runner.get("jockeyForm", "")).replace(".","").isdigit() else 0.2 odds_prob = float(runner.get("impliedOddsProbability", 0.25)) if str(runner.get("impliedOddsProbability", "")).replace(".","").isdigit() else 0.25 track_bias = 1.0 if barrier <= 4 else 0.85 features = [ (17 - barrier) / 16.0, (60.0 - weight) / 8.0, (rating - 65.0) / 50.0, 1.0 / (recent_finish + 0.5), track_match, dist_sr, trainer_form, jockey_form, odds_prob, track_bias ] runners_features.append(features) probabilities = model.predict_proba(np.array(runners_features))[:, 1] ranked = np.argsort(probabilities)[::-1] sel1 = runner_names[ranked[0]] sel2 = runner_names[ranked[1]] if len(runner_names) > 1 else "Runner B" sel3 = runner_names[ranked[2]] if len(runner_names) > 2 else "Runner C" report.append(f"- **1st (Win):** {sel1} (Prob: {probabilities[ranked[0]]:.3f})") report.append(f"- **2nd (Place):** {sel2} (Prob: {probabilities[ranked[1]]:.3f})") report.append(f"- **3rd (Show):** {sel3} (Prob: {probabilities[ranked[2]]:.3f})") report.append(f"- *Recommended Box Trifecta:* ({sel1} / {sel2} / {sel3})\\n") else: report.append("- *Runner data unavailable for this race.*\\n") return "\\n".join(report) with gr.Blocks(theme=gr.themes.Soft()) as demo: gr.Markdown("# 🐎 Trifecta Pro + FormFav Racing Intelligence") gr.Markdown("Auto-fetch all Australian gallops meetings and generate optimized ML box trifecta predictions.") with gr.Tab("Auto-Fetch All Australian Gallops"): with gr.Row(): api_key_input = gr.Textbox(label="FormFav API Key", type="password", placeholder="Enter your FormFav API key...") date_input = gr.Textbox(label="Date (YYYY-MM-DD)", value="2026-10-30") fetch_btn = gr.Button("🚀 Get All Aus Gallops & Predict Trifectas", variant="primary") output_markdown = gr.Markdown(label="Generated Trifecta Predictions Report") fetch_btn.click( fn=fetch_all_aus_meetings_and_predict, inputs=[api_key_input, date_input], outputs=output_markdown ) if __name__ == "__main__": demo.launch() """ (space_dir / "app.py").write_text(app_py_content) model_path = Path("trifecta_pro_optimized_model.joblib") if model_path.exists(): import shutil shutil.copy(model_path, space_dir / "trifecta_pro_optimized_model.joblib") api.upload_folder( folder_path=str(space_dir), repo_id=space_id, repo_type="space", token=token ) print(f"✅ Successfully updated Space with Auto-Fetch All Australian Gallops at https://huggingface.co/spaces/{space_id}") if __name__ == "__main__": push_model_to_hub() push_dataset_to_hub() deploy_hf_space() print("🎉 Auto-fetch integration complete!")