Repository navigation
Expand file tree
/
Copy pathapp.py
More file actions
2183 lines (1878 loc) · 110 KB
/
Copy pathapp.py
File metadata and controls
2183 lines (1878 loc) · 110 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
import os
import pickle
import re
import logging
import json
import time
import requests
import copy
from bs4 import BeautifulSoup
from collections import defaultdict
from tavily import TavilyClient
from requests.exceptions import HTTPError
from collections import defaultdict
from typing import List, Dict, Any, Tuple, Set, Optional
from concurrent.futures import ThreadPoolExecutor, as_completed
from abc import ABC, abstractmethod
# --- Data Handling Imports ---
import numpy as np
import pandas as pd
logger = logging.getLogger(__name__)
# --- Machine Learning Imports ---
import xgboost as xgb
try:
from sklearn.preprocessing import StandardScaler
except ImportError:
logging.error("Scikit-learn not installed. `pip install scikit-learn`")
StandardScaler = None
# --- LLM and API Imports ---
import google.generativeai as genai
from dotenv import load_dotenv
# --- Web Search Import ---
try:
from duckduckgo_search import DDGS
WEB_SEARCH_ENABLED = True
except ImportError:
logging.warning("duckduckgo-search not installed. Web search disabled. `pip install duckduckgo-search`")
DDGS = None
WEB_SEARCH_ENABLED = False
# --- Supabase Import ---
SUPABASE_CLIENT: Optional["Client"] = None
SUPABASE_ENABLED = False
try:
from supabase import create_client, Client
except ImportError:
logging.warning("supabase-py not installed. Database logging disabled. `pip install supabase`")
create_client = None
Client = None # Ensure Client type is not available if import fails
# Import logger functions - adjust import path if supabase_logger.py is not in the same directory
from supabase_logger import (
log_new_prediction_session,
update_prediction_session_analysis,
SUPABASE_PREDICTION_TABLE_NAME
)
# --- UI Imports ---
import gradio as gr
# --- Configuration and Setup ---
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# Load Environment Variables
load_dotenv()
# Get Environment Variables
API_KEY = os.getenv("GOOGLE_API_KEY")
SUPABASE_URL = os.getenv("SUPABASE_URL")
SUPABASE_SERVICE_KEY = os.getenv("SUPABASE_SERVICE_KEY")
# --- Configure Google Gemini API Client ---
GEMINI_MODEL_NAME = 'gemini-2.0-flash'
GEMINI_ENABLED = False
llm_model = None
if not API_KEY:
logging.error("GOOGLE_API_KEY environment variable not set. LLM features disabled.")
else:
try:
genai.configure(api_key=API_KEY)
global_generation_config = {
"temperature": 0.3,
"top_p": 0.8,
"top_k": 40,
"max_output_tokens": 14096,
}
safety_settings = [
{"category": "HARM_CATEGORY_HARASSMENT", "threshold": "BLOCK_MEDIUM_AND_ABOVE"},
{"category": "HARM_CATEGORY_HATE_SPEECH", "threshold": "BLOCK_MEDIUM_AND_ABOVE"},
{"category": "HARM_CATEGORY_SEXUALLY_EXPLICIT", "threshold": "BLOCK_MEDIUM_AND_ABOVE"},
{"category": "HARM_CATEGORY_DANGEROUS_CONTENT", "threshold": "BLOCK_MEDIUM_AND_ABOVE"},
]
try:
llm_model = genai.GenerativeModel(GEMINI_MODEL_NAME,
generation_config=genai.GenerationConfig(**global_generation_config),
safety_settings=safety_settings)
llm_model.count_tokens("hello world")
GEMINI_ENABLED = True
logging.info(f"Gemini configured successfully (Model: {GEMINI_MODEL_NAME}).")
except Exception as api_e:
logging.exception(f"Failed to initialize or test Gemini model {GEMINI_MODEL_NAME}. LLM features disabled.")
llm_model = None
GEMINI_ENABLED = False
except Exception as e:
logging.exception("Error configuring or initializing Gemini model:")
llm_model = None
GEMINI_ENABLED = False
# --- Configure Supabase Client ---
if SUPABASE_URL and SUPABASE_SERVICE_KEY and create_client:
try:
SUPABASE_CLIENT = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY)
SUPABASE_ENABLED = True
logging.info("Supabase client initialized successfully.")
except Exception as e:
logging.exception("Failed to initialize Supabase client. Database logging disabled.")
SUPABASE_CLIENT = None
SUPABASE_ENABLED = False
elif not SUPABASE_URL or not SUPABASE_SERVICE_KEY:
logging.warning("SUPABASE_URL or SUPABASE_SERVICE_KEY not set. Database logging disabled.")
SUPABASE_CLIENT = None
# --- Load Scaler and XGBoost Model ---
MODEL_DIR = "model"
SCALER_PATH = os.path.join(MODEL_DIR, "scaler.pkl")
MODEL_PATH_PKL = os.path.join(MODEL_DIR, "xgboost_model.pkl")
SCALER = None
XGB_MODEL = None
SCALER_LOADED = False
MODEL_LOADED = False
# Load Scaler
if StandardScaler:
try:
logging.info(f"Attempting to load scaler from: {SCALER_PATH}")
with open(SCALER_PATH, 'rb') as f:
SCALER = pickle.load(f)
if hasattr(SCALER, 'transform'):
SCALER_LOADED = True
logging.info(f"Scaler loaded successfully from {SCALER_PATH}")
else:
logging.error(f"Object loaded from {SCALER_PATH} is not a valid scaler.")
SCALER = None
except FileNotFoundError:
logging.error(f"Scaler file not found at {SCALER_PATH}")
except Exception as e:
logging.exception(f"An unexpected error occurred loading scaler from {SCALER_PATH}:")
# Load XGBoost Model
if SCALER_LOADED: # Only try loading model if scaler was successful
try:
logging.info(f"Attempting to load XGBoost model from pickle: {MODEL_PATH_PKL}")
with open(MODEL_PATH_PKL, 'rb') as f:
XGB_MODEL = pickle.load(f)
if hasattr(XGB_MODEL, 'predict_proba'):
MODEL_LOADED = True
logging.info(f"XGBoost model loaded successfully from Pickle: {MODEL_PATH_PKL} (has predict_proba)")
elif hasattr(XGB_MODEL, 'predict'):
MODEL_LOADED = True
logging.warning(f"XGBoost model loaded successfully from Pickle: {MODEL_PATH_PKL}, but missing 'predict_proba'. Probabilities cannot be generated.")
XGB_MODEL = None # Model must have predict_proba for this application
MODEL_LOADED = False
else:
logging.error(f"Object loaded from {MODEL_PATH_PKL} is not a valid XGBoost model.")
XGB_MODEL = None
MODEL_LOADED = False
except FileNotFoundError:
logging.error(f"XGBoost model file not found at {MODEL_PATH_PKL}")
except pickle.UnpicklingError as e:
logging.exception(f"Error unpickling model from {MODEL_PATH_PKL}. Version mismatch?")
except Exception as e:
logging.exception(f"An unexpected error occurred loading XGBoost model from Pickle {MODEL_PATH_PKL}:")
else:
logging.error("Scaler did not load successfully. Skipping model loading.")
# --- Constants ---
# Ensure these match your model training
EXPECTED_FEATURE_ORDER = ['W', 'D', 'L']
# Map model output indices to outcome codes
MODEL_OUTPUT_MAPPING = {0: 'D', 1: 'L', 2: 'W'} # Assuming 0=Draw, 1=Loss, 2=Win based on XGB default sorting
# Map outcome codes back to model output indices (for probability extraction)
PROBABILITY_MAPPING = {v: k for k, v in MODEL_OUTPUT_MAPPING.items()}
# --- Manual Cache for Web Search ---
search_cache = {}
CACHE_TTL_SECONDS = 3600 # Cache web search results for 1 hour
# --- Helper Function: NumPy float/int converter for JSON ---.
def convert_numpy_floats(obj):
"""Recursively converts NumPy floats/ints to standard Python types for JSON."""
if isinstance(obj, (np.float32, np.float64)):
return float(obj)
elif isinstance(obj, (np.int32, np.int64)):
return int(obj)
elif isinstance(obj, dict):
return {k: convert_numpy_floats(v) for k, v in obj.items()}
elif isinstance(obj, list):
return [convert_numpy_floats(i) for i in obj]
elif isinstance(obj, np.ndarray):
# Convert arrays to lists and recurse
return convert_numpy_floats(obj.tolist())
return obj
# --- Data Parsing and Formatting ---
def parse_odds_and_teams(text):
"""
Extracts odds (W, D, L) and potentially team names from input text.
Improved team and odds parsing.
Returns {'odds': {'W': float, 'D': float, 'L': float}, 'teams': (str, str) or None} or None.
"""
logging.debug(f"Attempting to parse odds and teams from: '{text}'")
parsed_data = {'odds': None, 'teams': None}
# Normalize whitespace and handle potential None input
cleaned_text = re.sub(r'\s+', ' ', text.strip()) if text else ""
if not cleaned_text:
return None
# --- Odds Parsing ---
odds = {}
# Try explicit pattern matching (H/Draw/A format with optional keys)
patterns_explicit = {
'W': r'(?:H(?:ome)?|Win)\s*[:=]?\s*(\d{1,4}(?:\.\d{1,3})?)',
'D': r'(?:Draw|X|D)\s*[:=]?\s*(\d{1,4}(?:\.\d{1,3})?)',
'L': r'(?:A(?:way)?|Loss)\s*[:=]?\s*(\d{1,4}(?:\.\d{1,3})?)'
}
found_explicit_odds = 0
for key, pattern in patterns_explicit.items():
match = re.search(pattern, cleaned_text, re.IGNORECASE)
if match:
try:
odds[key] = float(match.group(1))
found_explicit_odds += 1
except (ValueError, IndexError):
logging.warning(f"Failed to convert explicit odd for {key}: {match.group(1)}")
pass # Ignore and continue if a specific odd fails
# Check if we found exactly 3 explicit odds
if found_explicit_odds == 3:
logging.info(f"Parsed odds using explicit keys: {odds}")
parsed_data['odds'] = odds
else:
implicit_pattern = r'(\d{1,4}(?:\.\d{1,3})?)\s+(\d{1,4}(?:\.\d{1,3})?)\s+(\d{1,4}(?:\.\d{1,3})?)\s*$'
match_implicit = re.search(implicit_pattern, cleaned_text)
if match_implicit:
try:
# Map to W, D, L assuming the order is W D L after team names
w, d, l = map(float, match_implicit.groups())
# Basic validation: odds must be >= 1.0
if w >= 1.0 and d >= 1.0 and l >= 1.0:
odds = {'W': w, 'D': d, 'L': l}
logging.info(f"Parsed odds using implicit 'W D L' format: {odds}")
parsed_data['odds'] = odds
else:
logging.warning(f"Implicit odds invalid (< 1.0): W={w}, D={d}, L={l}")
except (ValueError, IndexError):
logging.warning("Implicit regex matched numbers, failed conversion to float.")
# If odds were successfully parsed, proceed to extract teams
if parsed_data['odds']:
# Extract the text *before* the matched odds pattern
text_before_odds = cleaned_text
if match_implicit:
text_before_odds = cleaned_text[:match_implicit.start()].strip()
elif found_explicit_odds == 3:
# Find the start of the *first* explicit odd pattern match to cut off the string
first_match_start = float('inf')
for pattern in patterns_explicit.values():
match = re.search(pattern, cleaned_text, re.IGNORECASE)
if match:
first_match_start = min(first_match_start, match.start())
if first_match_start != float('inf'):
text_before_odds = cleaned_text[:first_match_start].strip()
# --- Team Name Parsing ---
if text_before_odds:
team_separator_match = re.search(r'([A-Za-z0-9][\w\s\.\-\'&]*)\s+(?:vs\.?|v\.?|against|\-|@)\s+([A-Za-z0-9][\w\s\.\-\'&]*)$', text_before_odds, re.IGNORECASE)
# If no match, try the hyphen separator format (Team1 - Team2)
if not team_separator_match:
team_separator_match = re.search(r'([A-Za-z0-9][\w\s\.\-\'&]*?)\s+-\s+([A-Za-z0-9][\w\s\.\-\'&]*)$', text_before_odds, re.IGNORECASE)
if team_separator_match:
team1 = team_separator_match.group(1).strip()
team2 = team_separator_match.group(2).strip()
# Basic validation: ensure teams are not just numbers or very short
if len(team1) > 1 and len(team2) > 1 and not team1.isdigit() and not team2.isdigit():
parsed_data['teams'] = (team1, team2)
logging.info(f"Extracted teams via separator: Home='{team1}', Away='{team2}'")
if not parsed_data.get('teams') and text_before_odds:
logging.info(f"Could not extract valid teams from text before odds: '{text_before_odds}'. Text was: '{text_before_odds}'")
if parsed_data['odds']:
return parsed_data
else:
if cleaned_text:
logging.warning(f"Could not parse 3 distinct odds from text: '{cleaned_text}'")
return None
def format_input_for_scaler(odds_dict):
"""Formats the odds dictionary into the NumPy array expected by the SCALER, respecting EXPECTED_FEATURE_ORDER."""
if not odds_dict or len(odds_dict) != 3:
logging.error("Invalid odds_dict provided to format_input_for_scaler.")
return None
# Ensure keys exist and values are numeric
if not all(key in odds_dict and isinstance(odds_dict[key], (int, float)) for key in EXPECTED_FEATURE_ORDER):
logging.error(f"Odds dictionary is missing keys or values are not numeric. Expected {EXPECTED_FEATURE_ORDER}. Got {odds_dict}")
return None
try:
input_list = [odds_dict[feature] for feature in EXPECTED_FEATURE_ORDER]
input_array = np.array([input_list], dtype=float)
if input_array.shape != (1, len(EXPECTED_FEATURE_ORDER)):
logging.error(f"Formatted input array shape {input_array.shape} != (1, {len(EXPECTED_FEATURE_ORDER)}).")
return None
logging.debug(f"Formatted input array for scaler: {input_array.tolist()}")
return input_array
except Exception as e:
logging.exception("Error formatting input for Scaler:")
return None
def predict_outcome(raw_input_array):
"""
Scales input, gets prediction and probabilities from the XGBoost model (expects .pkl with predict_proba).
Returns: dict {'prediction': 'W'/'D'/'L', 'probabilities': {'W': float, 'D': float, 'L': float}} or None on error.
Probabilities will be standard Python floats after conversion for JSON metadata.
"""
if not SCALER_LOADED or not MODEL_LOADED or SCALER is None or XGB_MODEL is None:
logging.error("Prediction attempt failed: Scaler or Model not ready.")
return None
if raw_input_array is None:
logging.error("Prediction failed: Invalid raw input.")
return None
try:
# 1. Scale the input
scaled_input = SCALER.transform(raw_input_array)
logging.info(f"Input scaled: Raw={raw_input_array.tolist()}, Scaled={scaled_input.tolist()}")
# 2. Predict Probabilities (Assuming Pickle has predict_proba)
if not hasattr(XGB_MODEL, 'predict_proba'):
logging.error("Loaded XGBoost model object does not have 'predict_proba' method. Cannot generate probabilities.")
return None
prediction_probs_raw = XGB_MODEL.predict_proba(scaled_input) # Shape (1, n_classes)
logging.info(f"Raw model probabilities: {prediction_probs_raw.tolist()}")
if prediction_probs_raw.ndim > 1:
prediction_probs_flat = prediction_probs_raw[0]
else:
logging.warning("predict_proba returned 1D array, expected 2D. Trying to proceed.")
prediction_probs_flat = prediction_probs_raw
if len(prediction_probs_flat) != len(MODEL_OUTPUT_MAPPING):
logging.error(f"Model returned {len(prediction_probs_flat)} probabilities, but expected {len(MODEL_OUTPUT_MAPPING)} classes based on mapping.")
return None
# 3. Determine Predicted Class
predicted_class_index = np.argmax(prediction_probs_flat)
predicted_outcome_code = MODEL_OUTPUT_MAPPING.get(predicted_class_index)
if predicted_outcome_code is None:
# This means the argmax index was not in MODEL_OUTPUT_MAPPING - should not happen with valid model output
logging.error(f"Predicted class index '{predicted_class_index}' not found in MODEL_OUTPUT_MAPPING. Check model output vs mapping.")
return None # Fail if mapping doesn't work
# 4. Create Probabilities Dictionary mapped to W/D/L (Convert to standard floats for JSON compatibility)
probabilities = {}
for outcome_code in EXPECTED_FEATURE_ORDER: # Use EXPECTED_FEATURE_ORDER for dictionary keys
class_index = PROBABILITY_MAPPING.get(outcome_code) # Get the index for this outcome code
if class_index is not None and class_index < len(prediction_probs_flat):
probabilities[outcome_code] = float(prediction_probs_flat[class_index]) # Convert to standard float
else:
# Fallback if somehow mapping is incomplete or index is out of bounds
logging.warning(f"Could not map outcome code {outcome_code} to a class index or index {class_index} is out of bounds.")
probabilities[outcome_code] = 0.0 # Use 0.0 as standard float
# 5. Return Result
result = {
'prediction': predicted_outcome_code,
'probabilities': probabilities # Contains standard floats
}
logging.info(f"Prediction result calculated: {result}")
return result
except AttributeError as ae:
if 'predict_proba' in str(ae):
logging.error("AttributeError: Model loaded from pickle does not support 'predict_proba'.")
logging.exception("Prediction failed due to AttributeError:")
return None
except Exception as e:
logging.exception("An unexpected error occurred during scaling or prediction:")
return None
# --- Helper function to format search results for LLM prompt ---
def format_search_results_for_llm(results_list, max_snippet_length=400):
"""
Formats the list of search result dictionaries into a human-readable string
suitable for inclusion in an LLM prompt as contextual information.
"""
if not results_list:
return "No relevant web search results found."
formatted_text = "--- EXTERNAL CONTEXTUAL ANALYSIS ---\n"
formatted_text += "Synthesize information from these sources for your analysis:\n\n"
for i, result in enumerate(results_list):
title = result.get('title', 'No Title')
body = result.get('body', 'No Body Content')
href = result.get('href', 'N/A')
category = result.get('category', 'GENERAL')
source_quality = result.get('source_quality', 0.0)
temporal_relevance = result.get('temporal_relevance', 0.0)
detected_date = result.get('detected_date', 'N/A')
# Truncate body content to avoid excessive prompt length
snippet = body[:max_snippet_length] + ('...' if len(body) > max_snippet_length else '')
formatted_text += f"## Source {i+1}: {title}\n"
formatted_text += f"**URL:** {href}\n"
formatted_text += f"**Category:** {category} | **Quality:** {source_quality:.1f} | **Temporal:** {temporal_relevance:.1f} | **Date:** {detected_date}\n"
formatted_text += f"**Snippet:** {snippet}\n\n"
formatted_text += "--- END EXTERNAL CONTEXTUAL ANALYSIS ---\n"
return formatted_text
# Search
DEFAULT_RAG_CONFIG = {
'search': {
'tavily_quota': int(os.getenv("TAVILY_QUOTA", "1000")),
'google_quota': int(os.getenv("GOOGLE_QUOTA", "100")),
'google_api_key': os.getenv("GOOGLE_API_KEY_CS"),
'google_cse_id': os.getenv("GOOGLE_CSE_ID"),
'tavily_api_key': os.getenv("TAVILY_API_KEY"),
'default_max_results': 5,
'retry_attempts': 2,
'retry_delay': 2, # seconds
'google_timeout': 8, # seconds
'tavily_depth': "advanced" # or "basic"
},
'processing': {
'trusted_sources': {
'sofascore.com': 0.9, 'whoscored.com': 0.9, 'betexplorer.com': 0.9, 'fotmob.com': 0.85,
'transfermarkt.com': 0.8, 'fbref.com': 0.8, 'understat.com': 0.85, 'espn.com': 0.75,
'bbc.co.uk': 0.8, 'skysports.com': 0.75, 'goal.com': 0.7, 'theanalyst.com': 0.85,
'oddschecker.com': 0.65, 'nytimes.com': 0.7, 'theguardian.com': 0.75,
'lequipe.fr': 0.7, 'marca.com': 0.65, 'bild.de': 0.6
},
'evidence_categories': {
'FORM': ['recent form', 'results', 'performance', 'streak', 'last matches', 'wins losses draws'],
'H2H': ['head to head', 'h2h', 'previous meetings', 'history between'],
'INJURIES': ['injury', 'injured', 'fitness', 'unavailable', 'doubtful', 'suspension', 'ruled out', 'player status'],
'LINEUP': ['lineup', 'starting xi', 'team news', 'formation', 'expected lineup', 'squad'],
'STATS': ['statistics', 'xg', 'possession', 'shots', 'passing', 'tackles', 'fouls', 'cards', 'corners', 'metrics'],
'CONTEXT': ['league position', 'standings', 'motivation', 'importance', 'scenario', 'qualification', 'table'],
'VENUE': ['home advantage', 'away record', 'stadium', 'pitch', 'crowd', 'venue'],
'ODDS': ['odds movement', 'market sentiment', 'betting patterns', 'price shift', 'bookie', 'lines'],
'PREDICTION': ['prediction', 'expert pick', 'forecast', 'tip', 'preview', 'analysis', 'probability']
},
# Weights for combined score components (must sum to 1.0) - Tunable
'scoring_weights': {'source': 0.5, 'temporal': 0.4, 'category_match': 0.1}
},
'enrichment': {
'enabled': True,
'workers': 5, # Threads for parallel fetching
'timeout': 10, # seconds for fetching
'min_text_length': 300, # Min chars after extraction to consider content useful
'max_text_length': 10000, # Max chars to keep from full text
'skip_extensions': ['.pdf', '.doc', '.docx', '.ppt', '.pptx', '.zip', '.rar', '.mp4', '.mp3', '.jpg', '.png', '.gif', '.xml', '.json']
},
'caching': {
'search_cache_ttl': 300, # TTL for raw search results cache
'search_cache_size': 100,
'enrich_cache_ttl': 600, # TTL for enriched content cache
'enrich_cache_size': 50,
'analyzer_cache_ttl': 3600, # TTL for the final RAG output cache (Match analysis result)
'analyzer_cache_size': 64
},
'results': {
'total_limit': 15,
'enrich_count': 5 # How many top results to attempt to enrich
}
}
# --- Unified Cache Manager ---
class CacheManager:
"""Unified cache implementation with TTL, size limits (LRU approximation), and deepcopy."""
def __init__(self, ttl: int = 300, max_size: int = 100, name: str = "Cache"):
self.ttl = ttl
self.max_size = max_size
self._cache: Dict[Any, Any] = {}
self._timestamps: Dict[Any, float] = {}
self._access_order: List[Any] = []
self.name = name
logger.info(f"Initialized {self.name} with TTL={ttl}s, MaxSize={max_size}")
def get(self, key: Any) -> Optional[Any]:
"""Get item from cache if valid, updates access order."""
if key in self._cache:
if time.time() - self._timestamps.get(key, 0) < self.ttl:
try:
self._access_order.remove(key)
self._access_order.append(key)
logger.debug(f"[{self.name}] Cache hit for key {key!r}")
return copy.deepcopy(self._cache[key])
except ValueError:
logger.debug(f"[{self.name}] Cache key {key!r} disappeared from access order during access.")
self.delete(key)
return None
except Exception as e:
logger.warning(f"[{self.name}] Failed to deepcopy cache entry {key!r}: {e}. Returning shallow copy.", exc_info=False)
return self._cache[key]
else:
logger.debug(f"[{self.name}] Cache expired for key {key!r}")
self.delete(key)
logger.debug(f"[{self.name}] Cache miss for key {key!r}")
return None
def set(self, key: Any, value: Any):
"""Set item in cache, handling eviction if needed."""
if key in self._cache:
self.delete(key)
while len(self._cache) >= self.max_size and self._access_order:
oldest_key = self._access_order.pop(0)
if oldest_key in self._cache:
logger.debug(f"[{self.name}] Evicting oldest cache entry: {oldest_key!r}")
del self._cache[oldest_key]
del self._timestamps[oldest_key]
try:
self._cache[key] = copy.deepcopy(value)
except Exception as e:
logger.warning(f"[{self.name}] Failed to deepcopy value for caching key {key!r}: {e}. Storing shallow copy as fallback.", exc_info=False)
self._cache[key] = value
self._timestamps[key] = time.time()
self._access_order.append(key)
logger.debug(f"[{self.name}] Cache set for key {key!r}. Current size: {len(self)}")
def delete(self, key: Any):
"""Delete item from cache."""
if key in self._cache:
try:
del self._cache[key]
del self._timestamps[key]
self._access_order.remove(key)
logger.debug(f"[{self.name}] Cache deleted for key {key!r}. Remaining size: {len(self)}")
except ValueError:
logger.debug(f"[{self.name}] Cache key {key!r} already gone from access order list during deletion.")
except KeyError:
logger.debug(f"[{self.name}] Cache key {key!r} already gone from dicts during deletion.")
def clear(self):
"""Clear the entire cache."""
self._cache.clear()
self._timestamps.clear()
self._access_order.clear()
logger.info(f"[{self.name}] Cache cleared.")
def __len__(self):
return len(self._cache)
def __contains__(self, key):
return key in self._cache and time.time() - self._timestamps.get(key, 0) < self.ttl
# --- Search Provider Interface ---
class SearchProvider(ABC):
"""Defines a uniform interface for search backends."""
def __init__(self, config: Dict):
self.config = config.get('search', {})
self._enabled = False
self._quota_used = 0
self._quota_limit = self.config.get(f'{self.provider_name.lower()}_quota', float('inf')) or float('inf')
@property
@abstractmethod
def provider_name(self) -> str:
"""Returns the name of the provider (e.g., 'Google', 'Tavily')."""
pass
@abstractmethod
def _perform_search(self, query: str, max_results: int) -> Optional[List[Dict[str, str]]]:
"""
Performs the actual search API call.
Returns list of dicts {'href': str, 'title': str, 'body': str} on success (can be empty []).
Returns None on API/network/format failure.
"""
pass
def search(self, query: str, max_results: int) -> Optional[List[Dict[str, str]]]:
"""Wrapper to perform search and handle quota increment."""
if not self._enabled:
logger.debug(f"[{self.provider_name}] Search skipped: Provider not enabled.")
return None
if self._quota_used >= self._quota_limit:
logger.debug(f"[{self.provider_name}] Search skipped: Quota exhausted ({self._quota_used}/{self._quota_limit}).")
return None
self._quota_used += 1
logger.info(f"[{self.provider_name}] ({self._quota_used}/{self._quota_limit}) Attempting search for: '{query}'")
return self._perform_search(query, max_results)
def available(self) -> bool:
"""Checks if the provider is enabled (initialization successful)."""
return self._enabled
# --- Concrete Providers ---
class GoogleProvider(SearchProvider):
@property
def provider_name(self) -> str:
return "Google"
def __init__(self, config: Dict):
super().__init__(config)
self._api_key = self.config.get("google_api_key")
self._cse_id = self.config.get("google_cse_id")
self._timeout = self.config.get("google_timeout", DEFAULT_RAG_CONFIG['search']['google_timeout'])
if self._api_key and self._cse_id:
try:
test_url = f"https://www.googleapis.com/customsearch/v1?key={self._api_key}&cx={self._cse_id}&q=test&num=1"
response = requests.get(test_url, timeout=self._timeout)
response.raise_for_status()
self._enabled = True
logger.info(f"✓ {self.provider_name} API initialized successfully.")
except Exception as e:
logger.warning(f"✗ {self.provider_name} initialization failed: {e}.", exc_info=False)
else:
logger.warning(f"✗ {self.provider_name} API keys not found.")
def _perform_search(self, query: str, max_results: int) -> Optional[List[Dict[str, str]]]:
try:
url = f"https://www.googleapis.com/customsearch/v1"
params = {
'key': self._api_key,
'cx': self._cse_id,
'q': query,
'num': max_results,
'safe': 'active'
}
response = requests.get(url, params=params, timeout=self._timeout)
response.raise_for_status()
data = response.json()
items = data.get('items', [])
if not items:
logger.info(f"[{self.provider_name}] No search results found for '{query}'")
return []
results = []
for item in items:
snippet = item.get('snippet', '')
pagemap = item.get('pagemap', {})
metatags = pagemap.get('metatags', [])
best_snippet = snippet
for mt in metatags:
og_desc = mt.get('og:description', '')
desc = mt.get('description', '')
best_snippet = max(best_snippet, og_desc, desc, key=len)
results.append({
'href': item.get('link'), 'title': item.get('title', ''), 'body': best_snippet
})
return results
except requests.exceptions.Timeout:
logger.warning(f"[{self.provider_name}] Search timed out for '{query}'.", exc_info=False)
return None
except requests.exceptions.RequestException as e:
logger.warning(f"[{self.provider_name}] Search failed for '{query}': {e}.", exc_info=False)
return None
except Exception as e:
logger.error(f"[{self.provider_name}] Unexpected error during search for '{query}': {e}.", exc_info=True)
return None
class TavilyProvider(SearchProvider):
@property
def provider_name(self) -> str:
return "Tavily"
def __init__(self, config: Dict):
super().__init__(config)
self._api_key = self.config.get("tavily_api_key")
self._search_depth = self.config.get("tavily_depth", DEFAULT_RAG_CONFIG['search']['tavily_depth'])
if self._api_key:
try:
from tavily import TavilyClient
self._client = TavilyClient(api_key=self._api_key)
_ = self._client.search(query="test", max_results=1, search_depth="basic")
self._enabled = True
logger.info(f"✓ {self.provider_name} API initialized successfully.")
except ImportError:
logger.warning(f"✗ {self.provider_name} initialization failed: 'tavily' library not installed.", exc_info=False)
except Exception as e:
logger.warning(f"✗ {self.provider_name} initialization failed: {e}.", exc_info=False)
else:
logger.warning(f"✗ {self.provider_name} API key not found.")
def _perform_search(self, query: str, max_results: int) -> Optional[List[Dict[str, str]]]:
try:
tavily_response = self._client.search(
query=query, max_results=max_results, search_depth=self._search_depth
)
if isinstance(tavily_response, dict) and 'results' in tavily_response:
hits = tavily_response.get('results', [])
if not hits:
logger.info(f"[{self.provider_name}] No search results found for '{query}'")
return []
results = [
{'href': hit.get('url'), 'title': hit.get('title', ''), 'body': hit.get('content', '')}
for hit in hits if isinstance(hit, dict)
]
return results
else:
logger.warning(f"[{self.provider_name}] Unexpected response format for '{query}': {tavily_response}")
return None
except Exception as e:
logger.warning(f"[{self.provider_name}] Search failed for '{query}': {e}.", exc_info=False)
return None
class DuckDuckGoProvider(SearchProvider):
@property
def provider_name(self) -> str:
return "DuckDuckGo"
def __init__(self, config: Dict):
super().__init__(config)
try:
from duckduckgo_search import DDGS
self._client = DDGS()
self._enabled = True
logger.info(f"✓ {self.provider_name} Search initialized successfully")
except ImportError:
logger.warning(f"✗ {self.provider_name} initialization failed: 'duckduckgo-search' library not installed.", exc_info=False)
except Exception as e:
logger.warning(f"✗ {self.provider_name} initialization failed: {e}.", exc_info=False)
self._quota_limit = float('inf')
def available(self) -> bool:
return self._enabled
def _perform_search(self, query: str, max_results: int) -> Optional[List[Dict[str, str]]]:
try:
hits = list(self._client.text(query, region='wt-wt', max_results=max_results))[:max_results]
if not hits:
logger.info(f"[{self.provider_name}] No search results found for '{query}'")
return []
results = [
{'href': r.get('href'), 'title': r.get('title', ''), 'body': r.get('body', '')}
for r in hits if isinstance(r, dict)
]
return results
except Exception as e:
logger.warning(f"[{self.provider_name}] Search failed for '{query}': {e}.", exc_info=False)
return None
# --- Composite Client with Retries and Cache ---
class CompositeSearchClient:
"""Unified interface for search providers with fallback, retries, and cache."""
def __init__(self, config: Dict):
self.config = config
self._search_config = config.get('search', DEFAULT_RAG_CONFIG['search'])
self.providers = self._init_providers(config)
self.cache = CacheManager(
ttl=config.get('caching', {}).get('search_cache_ttl', DEFAULT_RAG_CONFIG['caching']['search_cache_ttl']),
max_size=config.get('caching', {}).get('search_cache_size', DEFAULT_RAG_CONFIG['caching']['search_cache_size']),
name="SearchClientCache"
)
self._retry_attempts = self._search_config.get("retry_attempts", DEFAULT_RAG_CONFIG['search']['retry_attempts'])
self._retry_delay = self._search_config.get("retry_delay", DEFAULT_RAG_CONFIG['search']['retry_delay'])
self._default_max_results = self._search_config.get("default_max_results", DEFAULT_RAG_CONFIG['search']['default_max_results'])
def _init_providers(self, config: Dict) -> List[SearchProvider]:
"""Initializes providers in preferred order (Google, Tavily, DDGS)."""
providers: List[SearchProvider] = []
google_prov = GoogleProvider(config)
if google_prov.available():
providers.append(google_prov)
tavily_prov = TavilyProvider(config)
if tavily_prov.available():
providers.append(tavily_prov)
ddgs_prov = DuckDuckGoProvider(config)
if ddgs_prov.available():
providers.append(ddgs_prov)
else:
pass
if not providers:
logger.error("No search providers successfully initialized. Search will always return empty.")
else:
logger.info(f"Initialized providers (in order): {[p.provider_name for p in providers]}")
return providers
def search(self, query: str, max_results: Optional[int] = None, force_refresh: bool = False) -> List[Dict]:
"""
Main search method with cascading fallbacks, retries, and caching.
Returns list of dicts {'href', 'title', 'body'}. Returns [] on failure.
"""
q = query.strip()
if not q:
logger.warning("Empty query provided to search client.")
return []
actual_max_results = max_results if max_results is not None else self._default_max_results
cache_key = (q, actual_max_results)
if not force_refresh:
cached = self.cache.get(cache_key)
if cached is not None:
return cached
logger.debug(f"SearchClientCache miss for query: '{q}' (max_results={actual_max_results}). Starting provider search...")
for provider in self.providers:
logger.debug(f"Trying {provider.provider_name} for '{q}'...")
attempt = 0
while attempt <= self._retry_attempts:
if not provider.available():
logger.debug(f"[{provider.provider_name}] Skipping attempt {attempt+1}: Provider not available or quota exhausted.")
break
try:
results = provider.search(q, actual_max_results)
if results is not None:
logger.debug(f"Search successful via {provider.provider_name} on attempt {attempt+1} for '{q}'")
self.cache.set(cache_key, results)
return results
else:
logger.warning(f"[{provider.provider_name}] Search returned None for '{q}' (attempt {attempt+1}/{self._retry_attempts})")
if attempt < self._retry_attempts:
time.sleep(self._retry_delay)
attempt += 1
else:
logger.error(f"[{provider.provider_name}] Failed after {self._retry_attempts+1} attempts for '{q}'. Trying next provider.")
break
except Exception as e:
logger.error(f"[{provider.provider_name}] Unexpected error DURING search attempt {attempt+1} for '{q}': {e}.", exc_info=True)
if attempt < self._retry_attempts:
time.sleep(self._retry_delay)
attempt += 1
else:
logger.error(f"[{provider.provider_name}] Failed after {self._retry_attempts+1} attempts with unexpected errors for '{q}'. Trying next provider.")
break
logger.error(f"All search providers failed after retries/fallbacks for query: '{q}'.")
empty_results: List[Dict] = []
self.cache.set(cache_key, empty_results)
return empty_results
# --- Query Builder ---
class QueryBuilder:
"""Constructs staged and targeted search queries based on match context and categories."""
def __init__(self, base_query: str, teams: Optional[List[str]], config: Dict):
self.config = config.get('processing', DEFAULT_RAG_CONFIG['processing'])
self.base_query = base_query.strip()
self._evidence_categories = self.config.get('evidence_categories', DEFAULT_RAG_CONFIG['processing']['evidence_categories'])
self._teams = teams if teams and len(teams) == 2 else None
self.team_str = self._build_team_string()
self.basic_templates = [
"{entity_string} match preview analysis",
"{entity_string} team news preview",
"{entity_string} prediction"
]
self.evidence_templates = {
'FORM': ["{entity_string} recent form analysis", "{entity_string} last 5 matches statistics"],
'H2H': ["{entity_string} head to head record", "{entity_string} previous meetings results"],
'INJURIES': ["{entity_string} injury news updates", "{entity_string} player availability fitness"],
'LINEUP': ["{entity_string} predicted lineup", "{entity_string} expected starting xi"],
'STATS': ["{entity_string} statistics xg analysis", "{teams_only_string} stats comparison"],
'CONTEXT': ["{entity_string} league context implications", "{entity_string} match importance"],
'VENUE': ["{entity_string} venue record", "{entity_string} stadium analysis"],
'ODDS': ["{entity_string} betting odds movement", "{entity_string} market trends"],
'PREDICTION': ["{entity_string} expert prediction", "{entity_string} betting tips"]
}
def _build_team_string(self) -> str:
"""Builds a string for query templates, prioritizing extracted teams."""
if self._teams:
return f"{self._teams[0]} vs {self._teams[1]}"
keywords_to_remove = r'\s*(?:recent|form|head|to|stats|analysis|betting|trends|odds|preview|match|injury|news|prediction|expert)\s*'
cleaned_query = re.sub(keywords_to_remove, ' ', self.base_query, flags=re.IGNORECASE).strip()
cleaned_query = re.sub(r'\s+', ' ', cleaned_query).strip()
return cleaned_query or self.base_query
def get_queries(self) -> Dict[str, List[Tuple[str, str]]]:
"""
Generates staged and categorized queries.
Returns: {'stage_name': [('query_string', 'category'), ...]}
"""
queries: Dict[str, List[Tuple[str, str]]] = {'basic': [], 'evidence': []}
teams_only_string = f"{self._teams[0]} vs {self._teams[1]}" if self._teams else self.team_str
for template in self.basic_templates:
query_str = template.format(entity_string=self.team_str)
queries['basic'].append((re.sub(r'\s+', ' ', query_str).strip(), 'GENERAL'))
for category, templates in self.evidence_templates.items():
for template in templates:
query_str = template.format(
entity_string=self.team_str,
teams_only_string=teams_only_string
)
queries['evidence'].append((re.sub(r'\s+', ' ', query_str).strip(), category))
unique_queries: Dict[str, List[Tuple[str, str]]] = {stage: list(set(q_list)) for stage, q_list in queries.items()}
logger.info(f"Generated {len(unique_queries['basic'])} basic queries and {len(unique_queries['evidence'])} evidence queries.")
return unique_queries
# --- Result Processor ---
class ResultProcessor:
"""Processes and scores raw search results, handles duplicates, and assigns categories."""
def __init__(self, config: Dict):
self.config = config.get('processing', DEFAULT_RAG_CONFIG['processing'])
self.trusted_sources = self.config.get('trusted_sources', DEFAULT_RAG_CONFIG['processing']['trusted_sources'])
self.evidence_categories = self.config.get('evidence_categories', DEFAULT_RAG_CONFIG['processing']['evidence_categories'])
self.scoring_weights = self.config.get('scoring_weights', DEFAULT_RAG_CONFIG['processing']['scoring_weights'])
self.seen_urls: Set[str] = set()
self.date_pattern = r'\b(?:Jan(?:uary)?|Feb(?:ruary)?|Mar(?:ch)?|Apr(?:il)?|May|Jun(?:e)?|Jul(?:y)?|Aug(?:ust)?|Sep(?:tember)?|Oct(?:ober)?|Nov(?:ember)?|Dec(?:ember)?)\s+\d{1,2}(?:st|nd|rd|th)?(?:\s*,?\s*\d{4})?\b|\b\d{1,2}[\/\-\.]\d{1,2}[\/\-\.]\d{2,4}\b|\b\d{4}[\/\-\.]\d{1,2}[\/\-\.]\d{1,2}\b|\b\d{1,2}(?:st|nd|rd|th)?\s+(?:Jan(?:uary)?|Feb(?:ruary)?|Mar(?:ch)?|Apr(?:il)?|May|Jun(?:e)?|Jul(?:y)?|Aug(?:ust)?|Sep(?:tember)?|Oct(?:ober)?|Nov(?:ember)?|Dec(?:ember)?)(?:\s*,?\s*\d{4})?\b'
def process_batch(self, results: List[Dict], query_tag: str, initial_category: str = 'GENERAL') -> List[Dict]:
"""Processes a batch of search results, adds scoring, categorization, filters duplicates."""
processed_results: List[Dict] = []
if not results:
logger.debug(f"[Processor] No results to process for query tag: {query_tag}")
return processed_results
for r in results:
url = r.get('href')
if not url:
logger.debug(f"[Processor] Skipping result with no URL from query tag: {query_tag}")
continue
normalized_url = self._normalize_url(url)
if normalized_url in self.seen_urls:
logger.debug(f"[Processor] Skipping duplicate URL: {url}")
continue
self.seen_urls.add(normalized_url)
result_data = {
'title': r.get('title', ''), 'body': r.get('body', ''),
'href': url, 'query_tag': query_tag, 'category': initial_category,
'source_quality': 0.0, 'temporal_relevance': 0.0, 'combined_score': 0.0
}
self._score_result(result_data)
self._categorize_result(result_data)
processed_results.append(result_data)
logger.debug(f"[Processor] Processed {len(processed_results)} new results from query tag: {query_tag}")
return processed_results
def _normalize_url(self, url: str) -> str:
"""Normalizes URL for duplicate checking."""
if not isinstance(url, str): return ""
normalized = re.sub(r'^https?://(?:www\.)?', '', url).rstrip('/')
return normalized
def _score_result(self, result: Dict):
"""Calculates and adds scoring metrics (source, temporal, combined)."""
url = result.get('href', '')
body = result.get('body', '')
title = result.get('title', '')
source_q = 0.5
domain_match = re.search(r'https?://(?:www\.)?([^/]+)', url)
if domain_match:
domain = domain_match.group(1)
source_q = self.trusted_sources.get(domain, 0.5)
result['source_quality'] = source_q
temporal_r = 0.1
combined_text_lower = (title + ' ' + body).lower()
if 'today' in combined_text_lower or 'yesterday' in combined_text_lower or re.search(r'\b\d+\s+(?:hour|minute)s?\s+ago', combined_text_lower):