Coverage for functions \ flipdare \ service \ payments \ risk_service.py: 70%

1#!/usr/bin/env python 

2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved. 

3# 

4# This file is part of Flipdare's proprietary software and contains 

5# confidential and copyrighted material. Unauthorised copying, 

6# modification, distribution, or use of this file is strictly 

7# prohibited without prior written permission from Flipdare Pty Ltd. 

8# 

9# This software includes third-party components licensed under MIT, 

10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details. 

11# 

12 

13from __future__ import annotations 

14from typing import TYPE_CHECKING 

15import math 

16import stripe 

17from flipdare.app_log import LOG 

18from flipdare.constants import ( 

19 HIGH_TRANSACTION_AMOUNT, 

20 IS_DEBUG, 

21 RISK_ANALYSIS_CHARGE_BACK_GRACE_CT, 

22 STRIPE_DASHBOARD_URL, 

23) 

24from flipdare.generated import ( 

25 RiskAssessmentModel, 

26 StripeAccountModel, 

27 PaymentDisputeLinkResponseSchema, 

28 AppFeeType, 

29 RiskFactor, 

30 RiskScore, 

31) 

32from flipdare.message.user_message import StripeMessage 

33from flipdare.payments.app_stripe_config import AppStripeConfig 

34from flipdare.service._service_provider import ServiceProvider 

35 

36if TYPE_CHECKING: 

37 from flipdare.manager.service_manager import ServiceManager 

38 from flipdare.manager.backend_manager import BackendManager 

39 from flipdare.manager.db_manager import DbManager 

40 

41__all__ = ["RiskService"] 

42 

43 

44class RiskService(ServiceProvider): 

45 

46 def __init__( 

47 self, 

48 stripe_client: stripe.StripeClient | None = None, 

49 stripe_config: AppStripeConfig | None = None, 

50 db_manager: DbManager | None = None, 

51 backend_manager: BackendManager | None = None, 

52 service_manager: ServiceManager | None = None, 

53 ) -> None: 

54 from flipdare.payments.app_stripe_proxy import create_stripe_proxy 

55 

56 self.proxy = create_stripe_proxy(stripe_config=stripe_config, stripe_client=stripe_client) 

57 

58 super().__init__( 

59 db_manager=db_manager, 

60 backend_manager=backend_manager, 

61 service_manager=service_manager, 

62 ) 

63 

64 def calculate_risk( 

65 self, 

66 settings: StripeAccountModel, 

67 fee_type: AppFeeType, 

68 transaction_amount: int | None = None, 

69 ) -> RiskAssessmentModel: 

70 """Calculate delay days based on account risk profile""" 

71 assesment = self._assess_account_risk( 

72 settings, 

73 is_vip=fee_type.is_vip, 

74 transaction_amount=transaction_amount, 

75 ) 

76 

77 if IS_DEBUG: 

78 score = assesment.risk_score 

79 days = score.days 

80 LOG().info( 

81 f"Risk assessment for account {settings.account_id}:" 

82 f"transaction amount {transaction_amount}, risk score {score} (days={days})\n{assesment}\n" 

83 ) 

84 

85 return assesment 

86 

87 def get_dispute_management_link( 

88 self, endpoint: str, settings: StripeAccountModel, dispute_id: str 

89 ) -> PaymentDisputeLinkResponseSchema: 

90 """Get dispute management options for connected accounts""" 

91 acct_type = settings.account_type 

92 account_id = settings.account_id 

93 

94 if acct_type.is_standard: 

95 return PaymentDisputeLinkResponseSchema( 

96 { 

97 "url": STRIPE_DASHBOARD_URL, 

98 "is_login_link": False, 

99 "dispute_url": STRIPE_DASHBOARD_URL + f"/disputes/{dispute_id}", 

100 "instructions": StripeMessage.DISPUTE_STANDARD_INSTRUCT, 

101 } 

102 ) 

103 

104 return self._get_express_dispute_link(endpoint, account_id) 

105 

106 # ======================================================================== 

107 # RISK HELPERS 

108 # ======================================================================== 

109 

110 def _assess_account_risk( 

111 self, 

112 settings: StripeAccountModel, 

113 is_vip: bool = False, 

114 transaction_amount: int | None = None, 

115 ) -> RiskAssessmentModel: 

116 """Assess risk factors for connected account""" 

117 high_transaction_amt = max(settings.highest_transaction_amount, transaction_amount or 0) 

118 high_transaction = high_transaction_amt > HIGH_TRANSACTION_AMOUNT 

119 

120 factors = { 

121 RiskFactor.DISPUTE_RATE: settings.disputed_rate, 

122 RiskFactor.REFUND_RATE: settings.refund_rate, 

123 RiskFactor.ACCOUNT_AGE_DAYS: settings.account_age_days, 

124 RiskFactor.TOTAL_VOLUME: settings.transaction_count, 

125 RiskFactor.TRANSACTION_COUNT: settings.transaction_count, 

126 RiskFactor.HIGH_AMOUNT_TRANSACTION: high_transaction, 

127 } 

128 score = 0.0 

129 

130 # 1. Chargebacks (The Priority) 

131 cb_count = factors[RiskFactor.DISPUTE_RATE] 

132 if cb_count <= RISK_ANALYSIS_CHARGE_BACK_GRACE_CT: 

133 score += cb_count * 10 # Linear minor penalty 

134 else: 

135 # Exponential spike after 2 

136 score += 20 + math.exp(cb_count - 2) * 15 

137 

138 # 2. Refunds (Linear) 

139 # Assume 5% is a "high" refund rate for scaling 

140 refund_rate = factors[RiskFactor.REFUND_RATE] 

141 score += (refund_rate / 0.05) * 20 

142 

143 # 3. Account Age (Linear Inverse) 

144 # New accounts are riskier. Penalty decreases until 90 days. 

145 age = factors[RiskFactor.ACCOUNT_AGE_DAYS] 

146 age_penalty = max( 

147 0, (90 - age) * 0.2 

148 ) # Max 18 points penalty for brand new accounts, linearly decreasing to 0 at 90 days 

149 score += age_penalty 

150 

151 # 4. Transaction Count (Trust Factor) 

152 # Fewer than 10 transactions adds risk 

153 tx_count = factors[RiskFactor.TRANSACTION_COUNT] 

154 tx_risk = max(0, (10 - tx_count) * 5) 

155 score += tx_risk 

156 

157 # 5. VIP Mitigation (Linear Reduction) 

158 if is_vip: 

159 # Reduces total risk by 30% linearly 

160 score = score * 0.7 

161 

162 normalized_score = min(score, 100) # Cap at 100 for normalization 

163 

164 if IS_DEBUG: 

165 msg = ( 

166 f"Risk Assessment Details:\n" 

167 f" --------------------------------------------------------\n" 

168 f" Overall Score: {normalized_score}\n" 

169 f" Age Penalty: {age_penalty:.2f} (Age={age} days)\n" 

170 f" Transaction Risk: {tx_risk:.2f}\n" 

171 f" --------------------------------------------------------\n" 

172 # f" Chargeback Count: {cb_count}\n" 

173 f" Refund Rate: {refund_rate:.2f}\n" 

174 f" High Transaction Amount: {high_transaction_amt}\n" 

175 f" --------------------------------------------------------\n" 

176 f" Risk Factors:\n" 

177 f"{'\n'.join(f' - {k}: {v}' for k, v in factors.items())}\n" 

178 ) 

179 LOG().debug(msg) 

180 

181 risk_score = RiskScore.from_score(normalized_score) 

182 return RiskAssessmentModel( 

183 overall_score=normalized_score, 

184 risk_score=risk_score, 

185 risk_factors=factors, 

186 ) 

187 

188 # ======================================================================== 

189 # DISPUTE HELPERS 

190 # ======================================================================== 

191 

192 def _get_express_dispute_link( 

193 self, endpoint: str, account_id: str 

194 ) -> PaymentDisputeLinkResponseSchema: 

195 """Create login link for Express account dispute management""" 

196 # Express accounts get limited dashboard access. You can create login links for them: 

197 try: 

198 login_link = self.proxy.create_login_link(endpoint, account_id) 

199 return PaymentDisputeLinkResponseSchema( 

200 { 

201 "is_login_link": True, 

202 "url": login_link, 

203 "instructions": "Use this link to access your Stripe Dashboard", 

204 } 

205 ) 

206 except Exception as e: 

207 LOG().error(f"Cannot create login link for account {account_id}: {e!s}") 

208 return PaymentDisputeLinkResponseSchema( 

209 { 

210 "is_login_link": False, 

211 "url": STRIPE_DASHBOARD_URL, 

212 "instructions": StripeMessage.DISPUTE_EXPRESS_INSTRUCT, 

213 } 

214 )