Coverage for src/pyrf_api/routes/market.py: 100%

1"""Market-related API routes.""" 

2 

3from fastapi import APIRouter, Depends 

4 

5from pyrf import AsianHandicap, BivariatePoisson, DixonColes, RingfinityModelZero, TotalGoals 

6from pyrf.core import GoalGrid 

7from pyrf.market import BetState, DecimalOdds, Market, Quote, QuotedLine 

8 

9from ..deps import verify_bearer_token 

10from ..schemas import ( 

11 CalibrateRequest, 

12 CalibrateResponse, 

13 ImpliedOddsResponse, 

14 ImpliedProbRequest, 

15 ImpliedProbResponse, 

16 QuoteLegSchema, 

17) 

18 

19router = APIRouter(prefix="/market", tags=["market"], dependencies=[Depends(verify_bearer_token)]) 

20 

21 

22def _surface(market: str, line: float): 

23 """Create a market surface from market type and line. 

24 

25 Args: 

26 market: Market type ("TotalGoals" or "AsianHandicap") 

27 line: The line value for the market 

28 

29 Returns: 

30 A TotalGoals or AsianHandicap surface instance 

31 

32 Raises: 

33 ValueError: If market type is not recognized 

34 """ 

35 if market == "TotalGoals": 

36 return TotalGoals(line=line) # type: ignore[invalid-argument-type] # attrs converter 

37 if market == "AsianHandicap": 

38 return AsianHandicap(line=line) # type: ignore[invalid-argument-type] # attrs converter 

39 raise ValueError(f"Unknown market: {market}") 

40 

41 

42def _model( 

43 model_type: str, 

44 lambda_home: float, 

45 lambda_away: float, 

46 rho: float | None, 

47) -> BivariatePoisson | DixonColes | RingfinityModelZero: 

48 """Create a goal-scoring model from type and parameters. 

49 

50 Args: 

51 model_type: Model type ("BivariatePoisson", "DixonColes", or "RingfinityModelZero") 

52 lambda_home: Expected goals for home team 

53 lambda_away: Expected goals for away team 

54 rho: Dixon-Coles correlation parameter (optional) 

55 

56 Returns: 

57 A goal-scoring model instance of the specified type 

58 

59 Raises: 

60 ValueError: If model type is not recognized 

61 """ 

62 base = BivariatePoisson(lambda_home=lambda_home, lambda_away=lambda_away) 

63 if model_type == "BivariatePoisson": 

64 return base 

65 if model_type == "DixonColes": 

66 r = rho if rho is not None else 0.0 

67 return DixonColes(dist=base, rho=r) 

68 if model_type == "RingfinityModelZero": 

69 return RingfinityModelZero(dist=base) 

70 raise ValueError(f"Unknown model: {model_type}") 

71 

72 

73@router.post("/implied-prob", response_model=ImpliedProbResponse) 

74def post_implied_prob(body: ImpliedProbRequest) -> ImpliedProbResponse: 

75 """Compute model-implied probabilities for a market/line.""" 

76 surface = _surface(body.market, body.line) 

77 model = _model(body.model, body.lambda_home, body.lambda_away, body.rho) 

78 grid = GoalGrid(n=body.n) if body.n is not None else GoalGrid.default() 

79 

80 probs = surface.implied_prob(model=model, grid=grid) 

81 out = {k.value: float(v) for k, v in probs.items()} 

82 return ImpliedProbResponse(probabilities=out) 

83 

84 

85@router.post("/implied-odds", response_model=ImpliedOddsResponse) 

86def post_implied_odds(body: ImpliedProbRequest) -> ImpliedOddsResponse: 

87 """Compute model-implied decimal odds for a market/line.""" 

88 surface = _surface(body.market, body.line) 

89 model = _model(body.model, body.lambda_home, body.lambda_away, body.rho) 

90 grid = GoalGrid(n=body.n) if body.n is not None else GoalGrid.default() 

91 

92 quote = surface.implied_odds(model=model, grid=grid) 

93 legs = [ 

94 QuoteLegSchema(side=quote.side_a.side.value, odds=quote.side_a.odds.value), 

95 QuoteLegSchema(side=quote.side_b.side.value, odds=quote.side_b.odds.value), 

96 ] 

97 return ImpliedOddsResponse( 

98 market=quote.side_a.market.value, 

99 line=quote.side_a.line.value, 

100 legs=legs, 

101 ) 

102 

103 

104@router.post("/calibrate", response_model=CalibrateResponse) 

105def post_calibrate(body: CalibrateRequest) -> CalibrateResponse: 

106 """Calibrate a Dixon-Coles model from Total Goals and Asian Handicap quotes.""" 

107 tg = Quote.build( 

108 market=Market.TOTAL_GOAL, 

109 line=QuotedLine(body.tg_quote.line), 

110 side_pair=(BetState.OVER, BetState.UNDR), 

111 odds_pair=(DecimalOdds(body.tg_quote.over_odds), DecimalOdds(body.tg_quote.under_odds)), 

112 ) 

113 ah = Quote.build( 

114 market=Market.ASIAN_HCAP, 

115 line=QuotedLine(body.ah_quote.line), 

116 side_pair=(BetState.HOME, BetState.AWAY), 

117 odds_pair=(DecimalOdds(body.ah_quote.home_odds), DecimalOdds(body.ah_quote.away_odds)), 

118 ) 

119 

120 cls: type[DixonColes] = RingfinityModelZero if body.model == "RingfinityModelZero" else DixonColes 

121 kwargs: dict[str, float] = {"xtol": body.xtol} 

122 if body.rho is not None and body.model != "RingfinityModelZero": 

123 kwargs["rho"] = body.rho 

124 

125 calibrated = cls.calibrate(tg_quote=tg, ah_quote=ah, **kwargs) 

126 

127 return CalibrateResponse( 

128 model=body.model, 

129 lambda_home=calibrated.dist.lambda_home, 

130 lambda_away=calibrated.dist.lambda_away, 

131 rho=calibrated.rho, 

132 )