Coverage for polar/metrics/metrics.py: 77%

527 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 16:17 +0000

1from collections import deque 1a

2from collections.abc import Iterable 1a

3from datetime import datetime 1a

4from enum import StrEnum 1a

5from typing import TYPE_CHECKING, ClassVar, Protocol, cast 1a

6 

7if TYPE_CHECKING: 7 ↛ 8line 7 didn't jump to line 8 because the condition on line 7 was never true1a

8 from .schemas import MetricsPeriod 

9 

10from sqlalchemy import ( 1a

11 ColumnElement, 

12 Float, 

13 Integer, 

14 SQLColumnExpression, 

15 case, 

16 func, 

17 or_, 

18 type_coerce, 

19) 

20 

21from polar.enums import SubscriptionRecurringInterval 1a

22from polar.kit.time_queries import TimeInterval 1a

23from polar.models import Checkout, Order, Subscription 1a

24from polar.models.checkout import CheckoutStatus 1a

25from polar.models.event import Event 1a

26from polar.models.subscription import CustomerCancellationReason 1a

27 

28from .queries import MetricQuery 1a

29 

30 

31class MetricType(StrEnum): 1a

32 scalar = "scalar" 1a

33 currency = "currency" 1a

34 currency_sub_cent = "currency_sub_cent" 1a

35 percentage = "percentage" 1a

36 

37 

38def cumulative_sum(periods: Iterable["MetricsPeriod"], slug: str) -> int | float: 1a

39 return sum(getattr(p, slug) for p in periods) 

40 

41 

42def cumulative_last(periods: Iterable["MetricsPeriod"], slug: str) -> int | float: 1a

43 dd = deque((getattr(p, slug) for p in periods), maxlen=1) 

44 return dd.pop() 

45 

46 

47class Metric(Protocol): 1a

48 slug: ClassVar[str] 1a

49 display_name: ClassVar[str] 1a

50 type: ClassVar[MetricType] 1a

51 

52 @classmethod 1a

53 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: ... 53 ↛ anywhereline 53 didn't jump anywhere: it always raised an exception.1a

54 

55 

56class SQLMetric(Metric, Protocol): 1a

57 query: ClassVar[MetricQuery] 1a

58 

59 @classmethod 1a

60 def get_sql_expression( 60 ↛ anywhereline 60 didn't jump anywhere: it always raised an exception.1a

61 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

62 ) -> ColumnElement[int] | ColumnElement[float]: ... 

63 

64 

65class MetaMetric(Metric, Protocol): 1a

66 @classmethod 1a

67 def compute_from_period(cls, period: "MetricsPeriod") -> int | float: ... 67 ↛ exitline 67 didn't return from function 'compute_from_period' because 1a

68 

69 

70class OrdersMetric(SQLMetric): 1a

71 slug = "orders" 1a

72 display_name = "Orders" 1a

73 type = MetricType.scalar 1a

74 query = MetricQuery.orders 1a

75 

76 @classmethod 1a

77 def get_sql_expression( 1a

78 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

79 ) -> ColumnElement[int]: 

80 return func.count(Order.id) 

81 

82 @classmethod 1a

83 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

84 return cumulative_sum(periods, cls.slug) 

85 

86 

87class RevenueMetric(SQLMetric): 1a

88 slug = "revenue" 1a

89 display_name = "Revenue" 1a

90 type = MetricType.currency 1a

91 query = MetricQuery.orders 1a

92 

93 @classmethod 1a

94 def get_sql_expression( 1a

95 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

96 ) -> ColumnElement[int]: 

97 return func.sum(Order.net_amount) 

98 

99 @classmethod 1a

100 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

101 return cumulative_sum(periods, cls.slug) 

102 

103 

104class NetRevenueMetric(SQLMetric): 1a

105 slug = "net_revenue" 1a

106 display_name = "Net Revenue" 1a

107 type = MetricType.currency 1a

108 query = MetricQuery.orders 1a

109 

110 @classmethod 1a

111 def get_sql_expression( 1a

112 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

113 ) -> ColumnElement[int]: 

114 return func.sum(Order.payout_amount) 

115 

116 @classmethod 1a

117 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

118 return cumulative_sum(periods, cls.slug) 

119 

120 

121class CumulativeRevenueMetric(SQLMetric): 1a

122 slug = "cumulative_revenue" 1a

123 display_name = "Cumulative Revenue" 1a

124 type = MetricType.currency 1a

125 query = MetricQuery.orders 1a

126 

127 @classmethod 1a

128 def get_sql_expression( 1a

129 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

130 ) -> ColumnElement[int]: 

131 return func.sum(Order.net_amount) 

132 

133 @classmethod 1a

134 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

135 return cumulative_last(periods, cls.slug) 

136 

137 

138class NetCumulativeRevenueMetric(SQLMetric): 1a

139 slug = "net_cumulative_revenue" 1a

140 display_name = "Net Cumulative Revenue" 1a

141 type = MetricType.currency 1a

142 query = MetricQuery.orders 1a

143 

144 @classmethod 1a

145 def get_sql_expression( 1a

146 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

147 ) -> ColumnElement[int]: 

148 return func.sum(Order.payout_amount) 

149 

150 @classmethod 1a

151 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

152 return cumulative_last(periods, cls.slug) 

153 

154 

155class AverageOrderValueMetric(SQLMetric): 1a

156 slug = "average_order_value" 1a

157 display_name = "Average Order Value" 1a

158 type = MetricType.currency 1a

159 query = MetricQuery.orders 1a

160 

161 @classmethod 1a

162 def get_sql_expression( 1a

163 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

164 ) -> ColumnElement[int]: 

165 return func.cast(func.ceil(func.avg(Order.net_amount)), Integer) 

166 

167 @classmethod 1a

168 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> float: 1a

169 total_orders = sum(getattr(p, "orders") for p in periods) 

170 revenue = sum(getattr(p, "revenue") for p in periods) 

171 return revenue / total_orders if total_orders > 0 else 0.0 

172 

173 

174class NetAverageOrderValueMetric(SQLMetric): 1a

175 slug = "net_average_order_value" 1a

176 display_name = "Net Average Order Value" 1a

177 type = MetricType.currency 1a

178 query = MetricQuery.orders 1a

179 

180 @classmethod 1a

181 def get_sql_expression( 1a

182 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

183 ) -> ColumnElement[int]: 

184 return func.cast(func.ceil(func.avg(Order.payout_amount)), Integer) 

185 

186 @classmethod 1a

187 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> float: 1a

188 total_orders = sum(getattr(p, "orders") for p in periods) 

189 revenue = sum(getattr(p, "net_revenue") for p in periods) 

190 return revenue / total_orders if total_orders > 0 else 0.0 

191 

192 

193class OneTimeProductsMetric(SQLMetric): 1a

194 slug = "one_time_products" 1a

195 display_name = "One-Time Products" 1a

196 type = MetricType.scalar 1a

197 query = MetricQuery.orders 1a

198 

199 @classmethod 1a

200 def get_sql_expression( 1a

201 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

202 ) -> ColumnElement[int]: 

203 return func.count(Order.id).filter(Order.subscription_id.is_(None)) 

204 

205 @classmethod 1a

206 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

207 return cumulative_sum(periods, cls.slug) 

208 

209 

210class OneTimeProductsRevenueMetric(SQLMetric): 1a

211 slug = "one_time_products_revenue" 1a

212 display_name = "One-Time Products Revenue" 1a

213 type = MetricType.currency 1a

214 query = MetricQuery.orders 1a

215 

216 @classmethod 1a

217 def get_sql_expression( 1a

218 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

219 ) -> ColumnElement[int]: 

220 return func.sum(Order.net_amount).filter(Order.subscription_id.is_(None)) 

221 

222 @classmethod 1a

223 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

224 return cumulative_sum(periods, cls.slug) 

225 

226 

227class OneTimeProductsNetRevenueMetric(SQLMetric): 1a

228 slug = "one_time_products_net_revenue" 1a

229 display_name = "One-Time Products Net Revenue" 1a

230 type = MetricType.currency 1a

231 query = MetricQuery.orders 1a

232 

233 @classmethod 1a

234 def get_sql_expression( 1a

235 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

236 ) -> ColumnElement[int]: 

237 return func.sum(Order.payout_amount).filter(Order.subscription_id.is_(None)) 

238 

239 @classmethod 1a

240 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

241 return cumulative_sum(periods, cls.slug) 

242 

243 

244class NewSubscriptionsMetric(SQLMetric): 1a

245 slug = "new_subscriptions" 1a

246 display_name = "New Subscriptions" 1a

247 type = MetricType.scalar 1a

248 query = MetricQuery.active_subscriptions 1a

249 

250 @classmethod 1a

251 def get_sql_expression( 1a

252 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

253 ) -> ColumnElement[int]: 

254 return func.count(Subscription.id).filter( 

255 i.sql_date_trunc( 

256 cast(SQLColumnExpression[datetime], Subscription.started_at) 

257 ) 

258 == i.sql_date_trunc(t) 

259 ) 

260 

261 @classmethod 1a

262 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

263 return cumulative_sum(periods, cls.slug) 

264 

265 

266class NewSubscriptionsRevenueMetric(SQLMetric): 1a

267 slug = "new_subscriptions_revenue" 1a

268 display_name = "New Subscriptions Revenue" 1a

269 type = MetricType.currency 1a

270 query = MetricQuery.orders 1a

271 

272 @classmethod 1a

273 def get_sql_expression( 1a

274 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

275 ) -> ColumnElement[int]: 

276 return func.sum(Order.net_amount).filter( 

277 i.sql_date_trunc( 

278 cast(SQLColumnExpression[datetime], Subscription.started_at) 

279 ) 

280 == i.sql_date_trunc(t) 

281 ) 

282 

283 @classmethod 1a

284 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

285 return cumulative_sum(periods, cls.slug) 

286 

287 

288class NewSubscriptionsNetRevenueMetric(SQLMetric): 1a

289 slug = "new_subscriptions_net_revenue" 1a

290 display_name = "New Subscriptions Net Revenue" 1a

291 type = MetricType.currency 1a

292 query = MetricQuery.orders 1a

293 

294 @classmethod 1a

295 def get_sql_expression( 1a

296 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

297 ) -> ColumnElement[int]: 

298 return func.sum(Order.payout_amount).filter( 

299 i.sql_date_trunc( 

300 cast(SQLColumnExpression[datetime], Subscription.started_at) 

301 ) 

302 == i.sql_date_trunc(t) 

303 ) 

304 

305 @classmethod 1a

306 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

307 return cumulative_sum(periods, cls.slug) 

308 

309 

310class RenewedSubscriptionsMetric(SQLMetric): 1a

311 slug = "renewed_subscriptions" 1a

312 display_name = "Renewed Subscriptions" 1a

313 type = MetricType.scalar 1a

314 query = MetricQuery.orders 1a

315 

316 @classmethod 1a

317 def get_sql_expression( 1a

318 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

319 ) -> ColumnElement[int]: 

320 return func.count(Subscription.id.distinct()).filter( 

321 i.sql_date_trunc( 

322 cast(SQLColumnExpression[datetime], Subscription.started_at) 

323 ) 

324 != i.sql_date_trunc(t) 

325 ) 

326 

327 @classmethod 1a

328 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

329 return cumulative_sum(periods, cls.slug) 

330 

331 

332class RenewedSubscriptionsRevenueMetric(SQLMetric): 1a

333 slug = "renewed_subscriptions_revenue" 1a

334 display_name = "Renewed Subscriptions Revenue" 1a

335 type = MetricType.currency 1a

336 query = MetricQuery.orders 1a

337 

338 @classmethod 1a

339 def get_sql_expression( 1a

340 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

341 ) -> ColumnElement[int]: 

342 return func.sum(Order.net_amount).filter( 

343 i.sql_date_trunc( 

344 cast(SQLColumnExpression[datetime], Subscription.started_at) 

345 ) 

346 != i.sql_date_trunc(t) 

347 ) 

348 

349 @classmethod 1a

350 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

351 return cumulative_sum(periods, cls.slug) 

352 

353 

354class RenewedSubscriptionsNetRevenueMetric(SQLMetric): 1a

355 slug = "renewed_subscriptions_net_revenue" 1a

356 display_name = "Renewed Subscriptions Net Revenue" 1a

357 type = MetricType.currency 1a

358 query = MetricQuery.orders 1a

359 

360 @classmethod 1a

361 def get_sql_expression( 1a

362 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

363 ) -> ColumnElement[int]: 

364 return func.sum(Order.payout_amount).filter( 

365 i.sql_date_trunc( 

366 cast(SQLColumnExpression[datetime], Subscription.started_at) 

367 ) 

368 != i.sql_date_trunc(t) 

369 ) 

370 

371 @classmethod 1a

372 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

373 return cumulative_sum(periods, cls.slug) 

374 

375 

376class ActiveSubscriptionsMetric(SQLMetric): 1a

377 slug = "active_subscriptions" 1a

378 display_name = "Active Subscriptions" 1a

379 type = MetricType.scalar 1a

380 query = MetricQuery.active_subscriptions 1a

381 

382 @classmethod 1a

383 def get_sql_expression( 1a

384 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

385 ) -> ColumnElement[int]: 

386 return func.count(Subscription.id) 

387 

388 @classmethod 1a

389 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

390 return cumulative_last(periods, cls.slug) 

391 

392 

393class MonthlyRecurringRevenueMetric(SQLMetric): 1a

394 slug = "monthly_recurring_revenue" 1a

395 display_name = "Monthly Recurring Revenue" 1a

396 type = MetricType.currency 1a

397 query = MetricQuery.active_subscriptions 1a

398 

399 @classmethod 1a

400 def get_sql_expression( 1a

401 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

402 ) -> ColumnElement[int]: 

403 return func.coalesce( 

404 func.sum( 

405 case( 

406 ( 

407 Subscription.recurring_interval 

408 == SubscriptionRecurringInterval.year, 

409 func.round(Subscription.amount / 12), 

410 ), 

411 ( 

412 Subscription.recurring_interval 

413 == SubscriptionRecurringInterval.month, 

414 Subscription.amount, 

415 ), 

416 ( 

417 Subscription.recurring_interval 

418 == SubscriptionRecurringInterval.week, 

419 func.round(Subscription.amount * 4), 

420 ), 

421 ( 

422 Subscription.recurring_interval 

423 == SubscriptionRecurringInterval.day, 

424 func.round(Subscription.amount * 30), 

425 ), 

426 ) 

427 ), 

428 0, 

429 ) 

430 

431 @classmethod 1a

432 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

433 return cumulative_last(periods, cls.slug) 

434 

435 

436class CommittedMonthlyRecurringRevenueMetric(SQLMetric): 1a

437 slug = "committed_monthly_recurring_revenue" 1a

438 display_name = "Committed Monthly Recurring Revenue" 1a

439 type = MetricType.currency 1a

440 query = MetricQuery.active_subscriptions 1a

441 

442 @classmethod 1a

443 def get_sql_expression( 1a

444 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

445 ) -> ColumnElement[int]: 

446 return func.coalesce( 

447 func.sum( 

448 case( 

449 ( 

450 Subscription.recurring_interval 

451 == SubscriptionRecurringInterval.year, 

452 func.round(Subscription.amount / 12), 

453 ), 

454 ( 

455 Subscription.recurring_interval 

456 == SubscriptionRecurringInterval.month, 

457 Subscription.amount, 

458 ), 

459 ( 

460 Subscription.recurring_interval 

461 == SubscriptionRecurringInterval.week, 

462 func.round(Subscription.amount * 4), 

463 ), 

464 ( 

465 Subscription.recurring_interval 

466 == SubscriptionRecurringInterval.day, 

467 func.round(Subscription.amount * 30), 

468 ), 

469 ) 

470 ).filter( 

471 or_( 

472 func.coalesce(Subscription.ended_at, Subscription.ends_at).is_( 

473 None 

474 ), 

475 i.sql_date_trunc( 

476 cast( 

477 SQLColumnExpression[datetime], 

478 func.coalesce(Subscription.ended_at, Subscription.ends_at), 

479 ) 

480 ) 

481 < i.sql_date_trunc(now), 

482 ) 

483 ), 

484 0, 

485 ) 

486 

487 @classmethod 1a

488 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

489 return cumulative_last(periods, cls.slug) 

490 

491 

492class CheckoutsMetric(SQLMetric): 1a

493 slug = "checkouts" 1a

494 display_name = "Checkouts" 1a

495 type = MetricType.scalar 1a

496 query = MetricQuery.checkouts 1a

497 

498 @classmethod 1a

499 def get_sql_expression( 1a

500 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

501 ) -> ColumnElement[int]: 

502 return func.count(Checkout.id) 

503 

504 @classmethod 1a

505 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

506 return cumulative_sum(periods, cls.slug) 

507 

508 

509class SucceededCheckoutsMetric(SQLMetric): 1a

510 slug = "succeeded_checkouts" 1a

511 display_name = "Succeeded Checkouts" 1a

512 type = MetricType.scalar 1a

513 query = MetricQuery.checkouts 1a

514 

515 @classmethod 1a

516 def get_sql_expression( 1a

517 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

518 ) -> ColumnElement[int]: 

519 return func.count(Checkout.id).filter( 

520 Checkout.status == CheckoutStatus.succeeded 

521 ) 

522 

523 @classmethod 1a

524 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

525 return cumulative_sum(periods, cls.slug) 

526 

527 

528class CheckoutsConversionMetric(SQLMetric): 1a

529 slug = "checkouts_conversion" 1a

530 display_name = "Checkouts Conversion Rate" 1a

531 type = MetricType.percentage 1a

532 query = MetricQuery.checkouts 1a

533 

534 @classmethod 1a

535 def get_sql_expression( 1a

536 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

537 ) -> ColumnElement[float]: 

538 return type_coerce( 

539 case( 

540 (func.count(Checkout.id) == 0, 0), 

541 else_=func.count(Checkout.id).filter( 

542 Checkout.status == CheckoutStatus.succeeded 

543 ) 

544 / func.count(Checkout.id), 

545 ), 

546 Float, 

547 ) 

548 

549 @classmethod 1a

550 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> float: 1a

551 total_checkouts = sum(getattr(p, "checkouts") for p in periods) 

552 total_succeeded = sum(getattr(p, "succeeded_checkouts") for p in periods) 

553 return total_succeeded / total_checkouts if total_checkouts > 0 else 0.0 

554 

555 

556class CanceledSubscriptionsMetric(SQLMetric): 1a

557 slug = "canceled_subscriptions" 1a

558 display_name = "Canceled Subscriptions" 1a

559 type = MetricType.scalar 1a

560 query = MetricQuery.canceled_subscriptions 1a

561 

562 @classmethod 1a

563 def get_sql_expression( 1a

564 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

565 ) -> ColumnElement[int]: 

566 return func.count(Subscription.id) 

567 

568 @classmethod 1a

569 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

570 return cumulative_sum(periods, cls.slug) 

571 

572 

573class CanceledSubscriptionsCustomerServiceMetric(SQLMetric): 1a

574 slug = "canceled_subscriptions_customer_service" 1a

575 display_name = "Canceled Subscriptions - Customer Service" 1a

576 type = MetricType.scalar 1a

577 query = MetricQuery.canceled_subscriptions 1a

578 

579 @classmethod 1a

580 def get_sql_expression( 1a

581 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

582 ) -> ColumnElement[int]: 

583 return func.count(Subscription.id).filter( 

584 Subscription.customer_cancellation_reason 

585 == CustomerCancellationReason.customer_service 

586 ) 

587 

588 @classmethod 1a

589 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

590 return cumulative_sum(periods, cls.slug) 

591 

592 

593class CanceledSubscriptionsLowQualityMetric(SQLMetric): 1a

594 slug = "canceled_subscriptions_low_quality" 1a

595 display_name = "Canceled Subscriptions - Low Quality" 1a

596 type = MetricType.scalar 1a

597 query = MetricQuery.canceled_subscriptions 1a

598 

599 @classmethod 1a

600 def get_sql_expression( 1a

601 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

602 ) -> ColumnElement[int]: 

603 return func.count(Subscription.id).filter( 

604 Subscription.customer_cancellation_reason 

605 == CustomerCancellationReason.low_quality 

606 ) 

607 

608 @classmethod 1a

609 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

610 return cumulative_sum(periods, cls.slug) 

611 

612 

613class CanceledSubscriptionsMissingFeaturesMetric(SQLMetric): 1a

614 slug = "canceled_subscriptions_missing_features" 1a

615 display_name = "Canceled Subscriptions - Missing Features" 1a

616 type = MetricType.scalar 1a

617 query = MetricQuery.canceled_subscriptions 1a

618 

619 @classmethod 1a

620 def get_sql_expression( 1a

621 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

622 ) -> ColumnElement[int]: 

623 return func.count(Subscription.id).filter( 

624 Subscription.customer_cancellation_reason 

625 == CustomerCancellationReason.missing_features 

626 ) 

627 

628 @classmethod 1a

629 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

630 return cumulative_sum(periods, cls.slug) 

631 

632 

633class CanceledSubscriptionsSwitchedServiceMetric(SQLMetric): 1a

634 slug = "canceled_subscriptions_switched_service" 1a

635 display_name = "Canceled Subscriptions - Switched Service" 1a

636 type = MetricType.scalar 1a

637 query = MetricQuery.canceled_subscriptions 1a

638 

639 @classmethod 1a

640 def get_sql_expression( 1a

641 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

642 ) -> ColumnElement[int]: 

643 return func.count(Subscription.id).filter( 

644 Subscription.customer_cancellation_reason 

645 == CustomerCancellationReason.switched_service 

646 ) 

647 

648 @classmethod 1a

649 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

650 return cumulative_sum(periods, cls.slug) 

651 

652 

653class CanceledSubscriptionsTooComplexMetric(SQLMetric): 1a

654 slug = "canceled_subscriptions_too_complex" 1a

655 display_name = "Canceled Subscriptions - Too Complex" 1a

656 type = MetricType.scalar 1a

657 query = MetricQuery.canceled_subscriptions 1a

658 

659 @classmethod 1a

660 def get_sql_expression( 1a

661 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

662 ) -> ColumnElement[int]: 

663 return func.count(Subscription.id).filter( 

664 Subscription.customer_cancellation_reason 

665 == CustomerCancellationReason.too_complex 

666 ) 

667 

668 @classmethod 1a

669 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

670 return cumulative_sum(periods, cls.slug) 

671 

672 

673class CanceledSubscriptionsTooExpensiveMetric(SQLMetric): 1a

674 slug = "canceled_subscriptions_too_expensive" 1a

675 display_name = "Canceled Subscriptions - Too Expensive" 1a

676 type = MetricType.scalar 1a

677 query = MetricQuery.canceled_subscriptions 1a

678 

679 @classmethod 1a

680 def get_sql_expression( 1a

681 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

682 ) -> ColumnElement[int]: 

683 return func.count(Subscription.id).filter( 

684 Subscription.customer_cancellation_reason 

685 == CustomerCancellationReason.too_expensive 

686 ) 

687 

688 @classmethod 1a

689 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

690 return cumulative_sum(periods, cls.slug) 

691 

692 

693class CanceledSubscriptionsUnusedMetric(SQLMetric): 1a

694 slug = "canceled_subscriptions_unused" 1a

695 display_name = "Canceled Subscriptions - Unused" 1a

696 type = MetricType.scalar 1a

697 query = MetricQuery.canceled_subscriptions 1a

698 

699 @classmethod 1a

700 def get_sql_expression( 1a

701 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

702 ) -> ColumnElement[int]: 

703 return func.count(Subscription.id).filter( 

704 Subscription.customer_cancellation_reason 

705 == CustomerCancellationReason.unused 

706 ) 

707 

708 @classmethod 1a

709 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

710 return cumulative_sum(periods, cls.slug) 

711 

712 

713class CanceledSubscriptionsOtherMetric(SQLMetric): 1a

714 slug = "canceled_subscriptions_other" 1a

715 display_name = "Canceled Subscriptions - Other" 1a

716 type = MetricType.scalar 1a

717 query = MetricQuery.canceled_subscriptions 1a

718 

719 @classmethod 1a

720 def get_sql_expression( 1a

721 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

722 ) -> ColumnElement[int]: 

723 return func.count(Subscription.id).filter( 

724 or_( 

725 Subscription.customer_cancellation_reason 

726 == CustomerCancellationReason.other, 

727 Subscription.customer_cancellation_reason.is_(None), 

728 ) 

729 ) 

730 

731 @classmethod 1a

732 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

733 return cumulative_sum(periods, cls.slug) 

734 

735 

736class ChurnedSubscriptionsMetric(SQLMetric): 1a

737 slug = "churned_subscriptions" 1a

738 display_name = "Churned Subscriptions" 1a

739 type = MetricType.scalar 1a

740 query = MetricQuery.churned_subscriptions 1a

741 

742 @classmethod 1a

743 def get_sql_expression( 1a

744 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

745 ) -> ColumnElement[int]: 

746 return func.count(Subscription.id) 

747 

748 @classmethod 1a

749 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

750 return cumulative_sum(periods, cls.slug) 

751 

752 

753class CostsMetric(SQLMetric): 1a

754 slug = "costs" 1a

755 display_name = "Costs" 1a

756 type = MetricType.currency_sub_cent 1a

757 query = MetricQuery.events 1a

758 

759 @classmethod 1a

760 def get_sql_expression( 1a

761 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

762 ) -> ColumnElement[int]: 

763 return func.sum( 

764 Event.user_metadata["_cost"]["amount"].as_numeric(17, 12) 

765 ).filter(Event.user_metadata["_cost"].is_not(None)) 

766 

767 @classmethod 1a

768 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

769 return cumulative_sum(periods, cls.slug) 

770 

771 

772class CumulativeCostsMetric(SQLMetric): 1a

773 slug = "cumulative_costs" 1a

774 display_name = "Cumulative Costs" 1a

775 type = MetricType.currency_sub_cent 1a

776 query = MetricQuery.events 1a

777 

778 @classmethod 1a

779 def get_sql_expression( 1a

780 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

781 ) -> ColumnElement[int]: 

782 return func.sum( 

783 Event.user_metadata["_cost"]["amount"].as_numeric(17, 12) 

784 ).filter(Event.user_metadata["_cost"].is_not(None)) 

785 

786 @classmethod 1a

787 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

788 return cumulative_last(periods, cls.slug) 

789 

790 

791class AverageRevenuePerUserMetric(SQLMetric): 1a

792 slug = "average_revenue_per_user" 1a

793 display_name = "Average Revenue Per User" 1a

794 type = MetricType.currency 1a

795 query = MetricQuery.active_subscriptions 1a

796 

797 @classmethod 1a

798 def get_sql_expression( 1a

799 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

800 ) -> ColumnElement[int]: 

801 return func.cast( 

802 case( 

803 (func.count(Subscription.customer_id.distinct()) == 0, 0), 

804 else_=func.coalesce( 

805 func.sum( 

806 case( 

807 ( 

808 Subscription.recurring_interval 

809 == SubscriptionRecurringInterval.year, 

810 func.round(Subscription.amount / 12), 

811 ), 

812 ( 

813 Subscription.recurring_interval 

814 == SubscriptionRecurringInterval.month, 

815 Subscription.amount, 

816 ), 

817 ( 

818 Subscription.recurring_interval 

819 == SubscriptionRecurringInterval.week, 

820 func.round(Subscription.amount * 4), 

821 ), 

822 ( 

823 Subscription.recurring_interval 

824 == SubscriptionRecurringInterval.day, 

825 func.round(Subscription.amount * 30), 

826 ), 

827 ) 

828 ), 

829 0, 

830 ) 

831 / func.count(Subscription.customer_id.distinct()), 

832 ), 

833 Integer, 

834 ) 

835 

836 @classmethod 1a

837 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 1a

838 return cumulative_last(periods, cls.slug) 

839 

840 

841class CostPerUserMetric(SQLMetric): 1a

842 slug = "cost_per_user" 1a

843 display_name = "Cost Per User" 1a

844 type = MetricType.currency_sub_cent 1a

845 query = MetricQuery.events 1a

846 

847 @classmethod 1a

848 def get_sql_expression( 1a

849 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

850 ) -> ColumnElement[float]: 

851 total_customers = func.count(func.distinct(Event.customer_id)) + func.count( 

852 func.distinct(Event.external_customer_id) 

853 ) 

854 

855 total_costs = func.sum( 

856 func.coalesce(Event.user_metadata["_cost"]["amount"].as_numeric(17, 12), 0) 

857 ) 

858 

859 return type_coerce( 

860 case( 

861 (total_customers == 0, 0), 

862 else_=total_costs / total_customers, 

863 ), 

864 Float, 

865 ) 

866 

867 @classmethod 1a

868 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> float: 1a

869 total_active_users = cumulative_last(periods, ActiveSubscriptionsMetric.slug) 

870 total_costs = sum(getattr(p, CostsMetric.slug) for p in periods) 

871 return total_costs / total_active_users if total_active_users > 0 else 0.0 

872 

873 

874class GrossMarginMetric(MetaMetric): 1a

875 slug = "gross_margin" 1a

876 display_name = "Gross Margin" 1a

877 type = MetricType.currency 1a

878 

879 @classmethod 1a

880 def compute_from_period(cls, period: "MetricsPeriod") -> float: 1a

881 revenue = period.cumulative_revenue 

882 costs = period.cumulative_costs 

883 return revenue - costs 

884 

885 @classmethod 1a

886 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> float: 1a

887 return cumulative_last(periods, cls.slug) 

888 

889 

890class GrossMarginPercentageMetric(MetaMetric): 1a

891 slug = "gross_margin_percentage" 1a

892 display_name = "Gross Margin %" 1a

893 type = MetricType.percentage 1a

894 

895 @classmethod 1a

896 def compute_from_period(cls, period: "MetricsPeriod") -> float: 1a

897 revenue = period.cumulative_revenue 

898 costs = period.cumulative_costs 

899 return (revenue - costs) / revenue if revenue > 0 else 0.0 

900 

901 @classmethod 1a

902 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> float: 1a

903 return cumulative_last(periods, cls.slug) 

904 

905 

906class CashflowMetric(MetaMetric): 1a

907 slug = "cashflow" 1a

908 display_name = "Cashflow" 1a

909 type = MetricType.currency 1a

910 

911 @classmethod 1a

912 def compute_from_period(cls, period: "MetricsPeriod") -> float: 1a

913 revenue = period.revenue 

914 costs = period.costs 

915 return revenue - costs 

916 

917 @classmethod 1a

918 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> float: 1a

919 return cumulative_sum(periods, cls.slug) 

920 

921 

922class ChurnRateMetric(MetaMetric): 1a

923 slug = "churn_rate" 1a

924 display_name = "Churn Rate" 1a

925 type = MetricType.percentage 1a

926 

927 @classmethod 1a

928 def compute_from_period(cls, period: "MetricsPeriod") -> float: 1a

929 active_during = period.active_subscriptions 

930 new = period.new_subscriptions 

931 churned = period.churned_subscriptions 

932 canceled = period.canceled_subscriptions 

933 active_at_start = active_during - new + churned 

934 return canceled / active_at_start if active_at_start > 0 else 0.0 

935 

936 @classmethod 1a

937 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> float: 1a

938 return cumulative_last(periods, cls.slug) 

939 

940 

941class LTVMetric(MetaMetric): 1a

942 slug = "ltv" 1a

943 display_name = "Lifetime Value" 1a

944 type = MetricType.currency 1a

945 

946 @classmethod 1a

947 def compute_from_period(cls, period: "MetricsPeriod") -> int: 1a

948 arpu = period.average_revenue_per_user 

949 cost_per_user = period.cost_per_user 

950 churn_rate = period.churn_rate 

951 

952 if churn_rate == 0: 

953 return 0 

954 

955 net_revenue_per_user = arpu - cost_per_user 

956 ltv = int(net_revenue_per_user / churn_rate) 

957 

958 return max(0, ltv) 

959 

960 @classmethod 1a

961 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int: 1a

962 return int(cumulative_last(periods, cls.slug)) 

963 

964 

965class ActiveUserMetric(SQLMetric): 1a

966 slug = "active_user_by_event" 1a

967 display_name = "Active User (By event)" 1a

968 type = MetricType.scalar 1a

969 query = MetricQuery.events 1a

970 

971 @classmethod 1a

972 def get_sql_expression( 1a

973 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

974 ) -> ColumnElement[int]: 

975 return func.count(func.distinct(Event.customer_id)) + func.count( 

976 func.distinct(Event.external_customer_id) 

977 ) 

978 

979 @classmethod 1a

980 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int: 1a

981 return int(cumulative_last(periods, ActiveSubscriptionsMetric.slug)) 

982 

983 

984METRICS_SQL: list[type[SQLMetric]] = [ 1a

985 OrdersMetric, 

986 RevenueMetric, 

987 NetRevenueMetric, 

988 CumulativeRevenueMetric, 

989 NetCumulativeRevenueMetric, 

990 CostsMetric, 

991 CumulativeCostsMetric, 

992 AverageOrderValueMetric, 

993 NetAverageOrderValueMetric, 

994 AverageRevenuePerUserMetric, 

995 CostPerUserMetric, 

996 ActiveUserMetric, 

997 OneTimeProductsMetric, 

998 OneTimeProductsRevenueMetric, 

999 OneTimeProductsNetRevenueMetric, 

1000 NewSubscriptionsMetric, 

1001 NewSubscriptionsRevenueMetric, 

1002 NewSubscriptionsNetRevenueMetric, 

1003 RenewedSubscriptionsMetric, 

1004 RenewedSubscriptionsRevenueMetric, 

1005 RenewedSubscriptionsNetRevenueMetric, 

1006 ActiveSubscriptionsMetric, 

1007 MonthlyRecurringRevenueMetric, 

1008 CommittedMonthlyRecurringRevenueMetric, 

1009 CheckoutsMetric, 

1010 SucceededCheckoutsMetric, 

1011 CheckoutsConversionMetric, 

1012 CanceledSubscriptionsMetric, 

1013 CanceledSubscriptionsCustomerServiceMetric, 

1014 CanceledSubscriptionsLowQualityMetric, 

1015 CanceledSubscriptionsMissingFeaturesMetric, 

1016 CanceledSubscriptionsSwitchedServiceMetric, 

1017 CanceledSubscriptionsTooComplexMetric, 

1018 CanceledSubscriptionsTooExpensiveMetric, 

1019 CanceledSubscriptionsUnusedMetric, 

1020 CanceledSubscriptionsOtherMetric, 

1021 ChurnedSubscriptionsMetric, 

1022] 

1023 

1024METRICS_POST_COMPUTE: list[type[MetaMetric]] = [ 1a

1025 ChurnRateMetric, 

1026 LTVMetric, 

1027 GrossMarginMetric, 

1028 GrossMarginPercentageMetric, 

1029 CashflowMetric, 

1030] 

1031 

1032METRICS: list[type[Metric]] = [ 1a

1033 *METRICS_SQL, 

1034 *METRICS_POST_COMPUTE, 

1035] 

1036 

1037__all__ = [ 1a

1038 "MetricType", 

1039 "Metric", 

1040 "SQLMetric", 

1041 "MetaMetric", 

1042 "METRICS_SQL", 

1043 "METRICS_POST_COMPUTE", 

1044 "METRICS", 

1045]