File size: 3,268 Bytes
5feba25 | 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 | import logging
from typing import List, Dict, Any
from app.config import FEATURE_RANGES, DEFAULT_MODEL_FEATURES
logger = logging.getLogger(__name__)
def validate_feature(name: str, value: Any) -> Any:
"""
Validate a single feature value against its constraints.
Args:
name: Feature name
value: Value to validate
Returns:
Validated value or None if invalid
"""
if value is None:
return None
if name not in FEATURE_RANGES:
return None
min_val, max_val, expected_type = FEATURE_RANGES[name]
# Convert to expected type
try:
converted = expected_type(value)
except (ValueError, TypeError):
logger.warning(f"Could not convert {name}={value} to {expected_type}")
return None
# Check range
if not (min_val <= converted <= max_val):
logger.warning(f"{name}={converted} is outside valid range [{min_val}, {max_val}]")
return None
return converted
def prepare_features_dict(state: Dict[str, Any]) -> Dict[str, Any]:
"""
Validate and prepare features dictionary.
Args:
state: Dictionary of features from memory
Returns:
Dictionary with validated features
"""
validated = {}
for feature in DEFAULT_MODEL_FEATURES:
value = state.get(feature)
validated[feature] = validate_feature(feature, value)
return validated
def prepare_feature_vector(state: Dict[str, Any]) -> List[float]:
"""
Convert state dict to feature vector for ML model.
The vector must have features in the exact order expected by the model.
Features are ordered as in DEFAULT_MODEL_FEATURES.
Missing values are filled with 0.0 (neutral value).
Args:
state: Dictionary with feature values from memory
Returns:
List of 16 floats ready for ML model prediction
"""
# Define feature order (MUST match training data order)
feature_order = DEFAULT_MODEL_FEATURES
vector = []
for feature_name in feature_order:
value = state.get(feature_name)
if value is not None:
try:
vector.append(float(value))
except (ValueError, TypeError):
logger.warning(f"Could not convert {feature_name}={value} to float, using 0.0")
vector.append(0.0)
else:
# Use 0.0 for missing values (neutral/default)
vector.append(0.0)
assert len(vector) == 16, f"Expected 16 features, got {len(vector)}"
return vector
def count_collected_features(state: Dict[str, Any]) -> int:
"""
Count how many features have been collected (non-null).
Args:
state: Dictionary with feature values
Returns:
Count of non-null features
"""
return sum(1 for v in state.values() if v is not None)
def is_ready_for_prediction(state: Dict[str, Any], min_features: int = 14) -> bool:
"""
Check if enough features collected for reliable prediction.
Args:
state: Dictionary with feature values
min_features: Minimum features needed (default 14/16)
Returns:
True if ready for prediction
"""
collected = count_collected_features(state)
return collected >= min_features
|