Coverage for packages / core / ratings / track_conditions.py: 0%

1"""Track condition adjustments for Elo ratings. 

2 

3Foundational stub for a model that learns how different track conditions 

4(wet, dry, heavy, fast) affect horse/driver/trainer performance. 

5 

6The core idea: 

7 - Some horses perform significantly better on wet tracks ("mudders") while 

8 others prefer fast, dry surfaces. 

9 - We maintain per-horse track condition adjustments that are learned from 

10 historical performance residuals, similar to how barrier/handicap 

11 adjustments are learned. 

12 - These adjustments are applied to the effective rating: a horse with a 

13 positive wet-track adjustment gets a rating boost when the track is heavy. 

14 

15This module is **not yet integrated** into the main ``RatingEngine``. 

16Integration would require: 

17 1. Adding a ``TrackConditionAdjustment`` model (DB table). 

18 2. Loading adjustments into the engine via a repository. 

19 3. Applying the adjustment in ``compute_effective_rating`` when a race's 

20 ``track_condition`` field is populated. 

21""" 

22 

23from __future__ import annotations 

24 

25from dataclasses import dataclass, field 

26from enum import StrEnum 

27 

28 

29class TrackConditionCategory(StrEnum): 

30 """Broad categorisation of track conditions. 

31 

32 Mapped from raw track condition strings (e.g. "Good3", "Heavy10", 

33 "Soft5", "Fast") into these categories for adjustment learning. 

34 """ 

35 

36 FAST = "fast" # Firm / fast / hard 

37 GOOD = "good" # Good / dead 

38 SOFT = "soft" # Soft / slow / easy 

39 HEAVY = "heavy" # Heavy / slop / muddy 

40 UNKNOWN = "unknown" 

41 

42 

43# ── Mapping helpers ────────────────────────────────────────────────────── 

44 

45 

46def _categorise_track(condition: str | None) -> TrackConditionCategory: 

47 """Map a raw track condition string to a category. 

48 

49 Handles common NZ/AU track rating formats: 

50 - "Good3", "Good4" → ``GOOD`` 

51 - "Soft5", "Soft6" → ``SOFT`` 

52 - "Heavy8", "Heavy10" → ``HEAVY`` 

53 - "Fast", "Firm" → ``FAST`` 

54 - "Dead" → ``GOOD`` 

55 - "Slow" → ``SOFT`` 

56 """ 

57 if not condition: 

58 return TrackConditionCategory.UNKNOWN 

59 

60 c = condition.strip().lower() 

61 

62 for keyword, cat in [ 

63 ("heavy", TrackConditionCategory.HEAVY), 

64 ("soft", TrackConditionCategory.SOFT), 

65 ("slow", TrackConditionCategory.SOFT), 

66 ("slop", TrackConditionCategory.HEAVY), 

67 ("muddy", TrackConditionCategory.HEAVY), 

68 ("good", TrackConditionCategory.GOOD), 

69 ("dead", TrackConditionCategory.GOOD), 

70 ("fast", TrackConditionCategory.FAST), 

71 ("firm", TrackConditionCategory.FAST), 

72 ("hard", TrackConditionCategory.FAST), 

73 ]: 

74 if keyword in c: 

75 return cat 

76 

77 return TrackConditionCategory.UNKNOWN 

78 

79 

80# ── Track condition model ──────────────────────────────────────────────── 

81 

82 

83@dataclass 

84class TrackConditionModel: 

85 """Learn and apply track condition adjustments. 

86 

87 Maintains a dictionary of ``(entity_type, entity_id, condition_category)`` 

88 → adjustment value, learned from performance residuals. 

89 """ 

90 

91 adjustments: dict[tuple[str, int, str], float] = field(default_factory=dict) 

92 sample_counts: dict[tuple[str, int, str], int] = field(default_factory=dict) 

93 

94 # Learning parameters 

95 learning_rate: float = 0.1 

96 min_samples: int = 5 

97 max_adjustment: float = 50.0 

98 

99 # ── Learning ───────────────────────────────────────────────────── 

100 

101 def learn_from_performance( 

102 self, 

103 entity_type: str, 

104 entity_id: int, 

105 track_condition: str | None, 

106 performance_residual: float, 

107 ) -> None: 

108 """Update the adjustment for an entity + track condition. 

109 

110 ``performance_residual`` measures how much the entity outperformed 

111 (positive) or underperformed (negative) expectations, in rating 

112 points. This is analogous to the residual computed in 

113 ``RatingEngine.learn_adjustments_from_race``. 

114 

115 Args: 

116 entity_type: ``"horse"``, ``"driver"``, or ``"trainer"``. 

117 entity_id: Entity ID. 

118 track_condition: Raw track condition string (e.g. ``"Good3"``). 

119 performance_residual: Rating delta to attribute to the condition. 

120 """ 

121 category = _categorise_track(track_condition) 

122 if category == TrackConditionCategory.UNKNOWN: 

123 return 

124 

125 key = (entity_type, entity_id, category.value) 

126 

127 current = self.adjustments.get(key, 0.0) 

128 count = self.sample_counts.get(key, 0) 

129 

130 # Incremental moving average 

131 new_count = count + 1 

132 new_adjustment = current + self.learning_rate * (performance_residual - current) 

133 new_adjustment = max( 

134 -self.max_adjustment, min(self.max_adjustment, new_adjustment) 

135 ) 

136 

137 self.adjustments[key] = new_adjustment 

138 self.sample_counts[key] = new_count 

139 

140 # ── Application ─────────────────────────────────────────────────── 

141 

142 def get_adjustment( 

143 self, 

144 entity_type: str, 

145 entity_id: int, 

146 track_condition: str | None, 

147 ) -> float: 

148 """Get the track condition adjustment for an entity. 

149 

150 Returns 0.0 if there are insufficient samples. 

151 

152 Args: 

153 entity_type: ``"horse"``, ``"driver"``, or ``"trainer"``. 

154 entity_id: Entity ID. 

155 track_condition: Raw track condition string. 

156 

157 Returns: 

158 Adjustment value in rating points (positive = better on this 

159 surface, negative = worse). 

160 """ 

161 category = _categorise_track(track_condition) 

162 if category == TrackConditionCategory.UNKNOWN: 

163 return 0.0 

164 

165 key = (entity_type, entity_id, category.value) 

166 count = self.sample_counts.get(key, 0) 

167 

168 if count < self.min_samples: 

169 return 0.0 

170 

171 return self.adjustments.get(key, 0.0) 

172 

173 # ── Population-level stats ──────────────────────────────────────── 

174 

175 def get_population_adjustment( 

176 self, 

177 track_condition: str | None, 

178 ) -> float: 

179 """Get the average adjustment across all entities for a condition. 

180 

181 Useful as a fallback when no per-entity data is available. 

182 

183 Args: 

184 track_condition: Raw track condition string. 

185 

186 Returns: 

187 Population-average adjustment. 

188 """ 

189 category = _categorise_track(track_condition) 

190 if category == TrackConditionCategory.UNKNOWN: 

191 return 0.0 

192 

193 vals = [ 

194 adj 

195 for (_, _, cat), adj in self.adjustments.items() 

196 if cat == category.value 

197 ] 

198 if not vals: 

199 return 0.0 

200 return sum(vals) / len(vals) 

201 

202 def reset(self) -> None: 

203 """Clear all learned adjustments (for re-learning).""" 

204 self.adjustments.clear() 

205 self.sample_counts.clear() 

206 

207 

208# ── Example / test usage ───────────────────────────────────────────────── 

209 

210 

211def _demo() -> None: 

212 """Demonstrate track condition learning and application.""" 

213 model = TrackConditionModel(learning_rate=0.15, min_samples=3) 

214 

215 print("Track Condition Model — demo") 

216 print() 

217 

218 # Simulate a "mudder" horse that performs better on wet tracks 

219 print("Simulating a horse that performs well on heavy tracks...") 

220 for _ in range(10): 

221 # Residual: +20 rating points on heavy tracks 

222 model.learn_from_performance("horse", 42, "Heavy10", 20.0) 

223 # Residual: -5 on good tracks 

224 model.learn_from_performance("horse", 42, "Good3", -5.0) 

225 

226 print(f" Horse 42 on Heavy: {model.get_adjustment('horse', 42, 'Heavy10'):+.1f}") 

227 print(f" Horse 42 on Good: {model.get_adjustment('horse', 42, 'Good3'):+.1f}") 

228 print( 

229 f" Horse 42 on Fast: {model.get_adjustment('horse', 42, 'Fast'):+.1f} (no data)" 

230 ) 

231 print(f" Population heavy: {model.get_population_adjustment('Heavy10'):+.1f}") 

232 

233 

234if __name__ == "__main__": 

235 _demo()