Coverage for adrs/types.py: 68%

1import polars as pl

2from collections import OrderedDict

3from datetime import datetime, timedelta, timezone

4from urllib.parse import parse_qsl

5from pydantic import (

6 BaseModel,

7 ConfigDict,

8 GetCoreSchemaHandler,

9 model_serializer,

10 model_validator,

11)

12from pydantic_core import CoreSchema, core_schema

14from typing import Any, Self, TypedDict, cast, Required, Literal

17class Data(TypedDict):

18 start_time: Required[datetime]

21class SubscriptionResponse(TypedDict):

22 conn_id: str

23 success: bool

24 message: str

27CollectedDataType = Literal["snapshot", "delta"]

30class CollectedData(TypedDict):

31 topic: str

32 data: list[Data]

33 local_timestamp_ms: int

34 type: CollectedDataType

37Message = SubscriptionResponse | CollectedData

40class BoundedSet[T]:

41 def __init__(self, maxsize: int):

42 self.maxsize = maxsize

43 self._store = OrderedDict()

45 def add(self, item: T) -> bool:

46 """Returns True if item was new, False if it was a duplicate."""

47 if item in self._store:

48 return False

49 if len(self._store) >= self.maxsize:

50 self._store.popitem(last=False) # evict oldest

51 self._store[item] = None

52 return True

54 def __contains__(self, item: T):

55 return item in self._store

57 def __len__(self):

58 return len(self._store)

61def _normalise(df: pl.DataFrame) -> pl.DataFrame:

62 """Label start_time as UTC at millisecond precision, the storage convention."""

63 if "start_time" not in df.columns:

64 return df

65 return df.with_columns(

66 pl.col("start_time")

67 .dt.replace_time_zone(time_zone="UTC")

68 .dt.cast_time_unit(time_unit="ms")

69 )

72def _conform(df: pl.DataFrame, schema: pl.Schema) -> pl.DataFrame:

73 """Widen df to schema, filling columns it lacks with nulls."""

74 missing = [

75 pl.lit(None, dtype=dtype).alias(name)

76 for name, dtype in schema.items()

77 if name not in df.columns

78 ]

79 return (df.with_columns(missing) if missing else df).cast(schema).select(*schema)

82class SortedDataList:

83 """Candles for one topic, kept sorted by start_time and stored columnar.

85 Rows live in a polars DataFrame, not as Python dicts. That matters at fine

86 intervals: a year of 1m candles is ~25 MB as Arrow but ~240 MB once every

87 row is boxed into a dict, and the old implementation paid that cost

88 permanently while rebuilding the frame on every read.

90 `data` still hands back `list[Data]` for callers that want dicts, but it

91 materialises a fresh snapshot each time — mutating it does not write back.

92 Use `append`/`merge`/`replace_last`/`tail`/`filter_time_range` instead.

93 """

95 def __init__(self, datas: list[Data] | None = None):

96 self._df = (

97 _normalise(pl.DataFrame(datas, infer_schema_length=None))

98 if datas

99 else pl.DataFrame()

100 )

101

102 @property

103 def data(self) -> list[Data]:

104 """Snapshot of the rows as dicts. Costly — prefer `to_df()`."""

105 return cast(list[Data], self._df.to_dicts())

106

107 @data.setter

108 def data(self, datas: list[Data]):

109 self._df = (

110 _normalise(pl.DataFrame(datas, infer_schema_length=None))

111 if datas

112 else pl.DataFrame()

113 )

114

115 def append(self, value: Data):

116 incoming = _normalise(pl.DataFrame([value], infer_schema_length=None))

117 if not self._df.width:

118 self._df = incoming

119 return

120 self._df = pl.concat([self._df, incoming], how="diagonal_relaxed").sort(

121 "start_time"

122 )

123

124 def sort(self):

125 self._df = self._df.sort("start_time")

126

127 def merge(self, datas: list[Data]):

128 if not datas:

129 return

130 self.merge_df(pl.DataFrame(datas, infer_schema_length=None))

131

132 def merge_df(self, df: pl.DataFrame):

133 """Fold df in, keeping one row per start_time with df's values winning."""

134 df = _normalise(df)

135 if not df.width:

136 return

137 if not self._df.width:

138 self._df = df.sort("start_time")

139 return

140

141 schema = pl.concat(

142 [self._df.clear(), df.clear()], how="diagonal_relaxed"

143 ).schema

144 cols = [name for name in schema if name != "start_time"]

145

146 self._df = (

147 _conform(self._df, schema)

148 .join(_conform(df, schema), how="full", on="start_time", coalesce=True)

149 .select(

150 "start_time",

151 *[pl.coalesce(f"{name}_right", name).alias(name) for name in cols],

152 )

153 .sort("start_time")

154 )

155

156 def tail(self, n: int):

157 """Keep only the most recent n rows."""

158 self._df = self._df.tail(n)

159

160 def replace_last(self, value: Data):

161 """Overwrite the final row, e.g. when a candle is restated."""

162 if not self._df.height:

163 self.append(value)

164 return

165 self._df = self._df.head(self._df.height - 1)

166 self.append(value)

167

168 def filter_time_range(self, start_time: datetime, end_time: datetime):

169 """Keep rows in [start_time, end_time)."""

170 if not self._df.width:

171 return

172 self._df = self._df.filter(

173 pl.col("start_time").is_between(start_time, end_time, closed="left")

174 )

175

176 def first_start_time(self) -> datetime | None:

177 return self._df["start_time"][0] if self._df.height else None

178

179 def last_start_time(self) -> datetime | None:

180 return self._df["start_time"][-1] if self._df.height else None

181

182 def is_empty(self) -> bool:

183 return self._df.height == 0

184

185 def to_df(self) -> pl.DataFrame:

186 return self._df

187

188 @classmethod

189 def from_df(cls, df: pl.DataFrame):

190 obj = cls()

191 obj._df = _normalise(df)

192 return obj

193

194 def __len__(self):

195 return self._df.height

196

197 def __getitem__(self, key):

198 if isinstance(key, int):

199 return cast(Data, self._df.row(key, named=True))

200 return self.data[key]

201

202

203class Symbol:

204 _MAX_LEN = 24

205

206 def __init__(self, s: str) -> None:

207 if len(s) > self._MAX_LEN:

208 raise ValueError(f"Symbol exceeds max length of {self._MAX_LEN}: {s!r}")

209 self._s = s

210

211 @classmethod

212 def from_str(cls, s: str) -> "Symbol":

213 return cls(s)

214

215 def hash(self) -> int:

216 return sum(ord(c) for c in self._s)

217

218 def split(self) -> tuple[str, str] | None:

219 for i, c in enumerate(self._s):

220 if c in "-/_:":

221 return self._s[:i], self._s[i + 1 :]

222

223 s = self._s

224

225 if len(s) >= 5 and s[-5:] in (

226 "FDUSD",

227 "fdusd",

228 "USDTM",

229 "usdtm",

230 "USDCM",

231 "usdcm",

232 ):

233 return s[:-5], s[-5:]

234

235 if len(s) >= 4 and s[-4:] in ("USDT", "USDC", "usdt", "usdc"):

236 return s[:-4], s[-4:]

237

238 if len(s) >= 3 and s[-3:] in ("USD", "usd"):

239 return s[:-3], s[-3:]

240

241 return None

242

243 def __str__(self) -> str:

244 return self._s

245

246 def __repr__(self) -> str:

247 return self._s

248

249 def __eq__(self, other: object) -> bool:

250 if isinstance(other, Symbol):

251 return self._s == other._s

252 if isinstance(other, str):

253 return self._s == other

254 return False

255

256 def __hash__(self) -> int:

257 return self.hash()

258

259 def __copy__(self) -> "Symbol":

260 return self

261

262 def __deepcopy__(self, memo: dict) -> "Symbol":

263 memo[id(self)] = self

264 return self

265

266 def __reduce__(self) -> tuple:

267 return (Symbol, (self._s,))

268

269 @classmethod

270 def __get_pydantic_core_schema__(

271 cls, source_type: Any, handler: GetCoreSchemaHandler

272 ) -> CoreSchema:

273 return core_schema.no_info_plain_validator_function(

274 lambda v: cls(v) if isinstance(v, str) else v,

275 serialization=core_schema.to_string_ser_schema(),

276 )

277

278

279class TopicError(Exception):

280 pass

281

282

283class Topic(BaseModel):

284 model_config = ConfigDict(frozen=True)

285

286 provider: str

287 endpoint: str

288 params: dict[str, str]

289

290 def query_params(self) -> dict[str, str]:

291 return self.params

292

293 def query_params_str(self) -> str:

294 if not self.params:

295 return ""

296 return "&".join(f"{k}={v}" for k, v in sorted(self.params.items()))

297

298 def endpoint_with_query_params(self) -> str:

299 if not self.params:

300 return self.endpoint

301 query = "&".join(f"{k}={v}" for k, v in sorted(self.params.items()))

302 return f"{self.endpoint}?{query}"

303

304 def __eq__(self, other: object) -> bool:

305 if not isinstance(other, Topic):

306 return False

307 return (

308 self.provider == other.provider

309 and self.endpoint_with_query_params() == other.endpoint_with_query_params()

310 )

311

312 def __hash__(self) -> int:

313 return hash((self.provider, self.endpoint_with_query_params()))

314

315 def __repr__(self) -> str:

316 return f"Topic {{ provider: {self.provider!r}, endpoint: {self.endpoint!r}, params: {dict(sorted(self.params.items()))!r} }}"

317

318 def __str__(self) -> str:

319 return f"{self.provider}|{self.endpoint_with_query_params()}"

320

321 def __copy__(self) -> "Topic":

322 return self

323

324 def __deepcopy__(self, memo: dict[int, Any] | None = None) -> "Topic":

325 if memo is not None:

326 memo[id(self)] = self

327 return self

328

329 def __reduce__(self) -> tuple:

330 return (Topic.from_str, (str(self),))

331

332 @model_validator(mode="before")

333 @classmethod

334 def _from_str(cls, v: Any) -> Any:

335 if isinstance(v, str):

336 return cls.from_str(v).__dict__

337 if isinstance(v, dict) and "query_params" in v and "params" not in v:

338 v = {**v, "params": v["query_params"]}

339 return v

340

341 @model_serializer

342 def _to_str(self) -> str:

343 return str(self)

344

345 @classmethod

346 def from_str(cls, s: str) -> "Topic":

347 tokens = s.split("|")

348 if len(tokens) < 2:

349 raise TopicError("cannot find provider/endpoint in the topic")

350 if len(tokens) > 2:

351 raise TopicError("topic must be in the '{provider}|{endpoint}' format")

352

353 provider, endpoint_with_query_params = tokens

354

355 if "?" in endpoint_with_query_params:

356 endpoint, query_string = endpoint_with_query_params.split("?", 1)

357 else:

358 endpoint, query_string = endpoint_with_query_params, ""

359 params: dict[str, str] = dict(parse_qsl(query_string))

360

361 if provider == "cryptoquant":

362 endpoint = endpoint.replace("flow/", "flows/", 1)

363

364 if params.get("symbol") and endpoint == "candle":

365 symbol_str = params["symbol"]

366 parsed_symbol = Symbol.from_str(symbol_str)

367 result = parsed_symbol.split()

368 if result is None:

369 raise TopicError(f"invalid symbol '{symbol_str}'")

370 base, quote = result

371 if "bybit" in provider or "binance" in provider:

372 params["symbol"] = f"{base}{quote}"

373 elif "coinbase" in provider:

374 params["symbol"] = f"{base}-{quote}"

375

376 return cls(provider=provider, endpoint=endpoint, params=params)

377

378 def get_params(self, key: str) -> str:

379 if key not in self.params:

380 raise TopicError(f"key '{key}' not found in topic '{self}'")

381 return self.params[key]

382

383 def interval(self) -> timedelta | None:

384 _MINUTE = 60_000

385 _HOUR = 60 * _MINUTE

386 _DAY = 24 * _HOUR

387

388 def ms(n: int) -> timedelta:

389 return timedelta(milliseconds=n)

390

391 match self.provider:

392 case "cryptoquant":

393 key = "window"

394 duration_map = {

395 "min": ms(_MINUTE),

396 "block": ms(10 * _MINUTE),

397 "10min": ms(10 * _MINUTE),

398 "hour": ms(_HOUR),

399 "day": ms(_DAY),

400 }

401 case "glassnode":

402 key = "i"

403 duration_map = {

404 "10m": ms(10 * _MINUTE),

405 "1h": ms(_HOUR),

406 "24h": ms(24 * _HOUR),

407 "1w": ms(7 * _DAY),

408 "1month": ms(30 * _DAY),

409 }

410 case "amberdata":

411 match self.endpoint:

412 case "derivatives/analytics/futures-perpetuals/apr-basis/constant-maturities":

413 return ms(15 * _MINUTE)

414 case "derivatives/analytics/trades-flow/gamma-exposures-snapshots":

415 return ms(_HOUR)

416 case (

417 "derivatives/analytics/futures-perpetuals/realized-funding-rates-cumulated"

418 | "derivatives/analytics/futures-perpetuals/funding-rates"

419 ):

420 return ms(8 * _HOUR)

421 case (

422 "derivatives/analytics/trades-flow/decorated-trades"

423 | "derivatives/analytics/trades-flow/options-yields/tradfi"

424 | "derivatives/analytics/trades-flow/option-volumes/tradfi"

425 | "derivatives/analytics/realized-volatility/implied-vs-realized/tradfi"

426 | "derivatives/analytics/realized-volatility/cones/tradfi"

427 | "derivatives/analytics/realized-volatility/cones"

428 | "derivatives/analytics/volatility/svi-minutely"

429 | "derivatives/analytics/options-scanner/top-trades"

430 | "derivatives/analytics/trades-flow/gamma-exposures/normalized-usd"

431 | "derivatives/analytics/trades-flow/block-volumes"

432 ):

433 return ms(_MINUTE)

434 key = "timeInterval"

435 duration_map = {

436 "day": ms(_DAY),

437 "hour": ms(_HOUR),

438 "minute": ms(_MINUTE),

439 }

440 case "coinalyze":

441 key = "interval"

442 duration_map = {

443 "1min": ms(_MINUTE),

444 "5min": ms(5 * _MINUTE),

445 "15min": ms(15 * _MINUTE),

446 "30min": ms(30 * _MINUTE),

447 "1hour": ms(_HOUR),

448 "2hour": ms(2 * _HOUR),

449 "4hour": ms(4 * _HOUR),

450 "6hour": ms(6 * _HOUR),

451 "12hour": ms(12 * _HOUR),

452 "daily": ms(_DAY),

453 }

454 case _:

455 key = "interval"

456 duration_map = {

457 "1m": ms(_MINUTE),

458 "3m": ms(3 * _MINUTE),

459 "5m": ms(5 * _MINUTE),

460 "10m": ms(10 * _MINUTE),

461 "15m": ms(15 * _MINUTE),

462 "30m": ms(30 * _MINUTE),

463 "1h": ms(_HOUR),

464 "2h": ms(2 * _HOUR),

465 "4h": ms(4 * _HOUR),

466 "6h": ms(6 * _HOUR),

467 "12h": ms(12 * _HOUR),

468 "1d": ms(_DAY),

469 "3d": ms(3 * _DAY),

470 "1w": ms(7 * _DAY),

471 "1M": ms(30 * _DAY),

472 }

473

474 v = self.params.get(key)

475 if v is None:

476 return None

477 return duration_map.get(v)

478

479 def cron(self) -> str | None:

480 match self.provider:

481 case "cryptoquant":

482 match self.params.get("window"):

483 case "min":

484 return "6 * * * * *"

485 case "block":

486 return "1 */10 * * * *"

487 case "10min":

488 return "0 */13 * * * *"

489 case "hour" if "/market-data/" in self.endpoint:

490 return "0 6 * * * *"

491 case "hour":

492 return "0 15 * * * *"

493 case "day":

494 return "0 6 0 * * *"

495 case _:

496 return None

497 case "glassnode":

498 match self.params.get("i"):

499 case "10m":

500 return "0 */13 * * * *"

501 case "1h":

502 return "0 6 * * * *"

503 case "24h":

504 return "0 6 0 * * *"

505 case "1w":

506 return "0 0 0 * * 1"

507 case "1month":

508 return "0 0 0 1 * *"

509 case _:

510 return None

511 case "coinbase":

512 match self.params.get("interval"):

513 case "1m":

514 return "6 * * * * *"

515 case "5m":

516 return "25 */5 * * * *"

517 case "15m":

518 return "15 */16 * * * *"

519 case "1h":

520 return "0 7 * * * *"

521 case "6h":

522 return "0 6 */7 * * *"

523 case "1d":

524 return "0 7 0 * * *"

525 case _:

526 return None

527 case "amberdata":

528 match self.params.get("timeInterval"):

529 case "minute":

530 return "3 * * * * *"

531 case "hour":

532 return "0 3 * * * *"

533 case "day":

534 return "0 3 0 * * *"

535 case _:

536 return None

537 case "coinalyze":

538 match self.params.get("interval"):

539 case "1min":

540 return "3 * * * * *"

541 case "5min":

542 return "3 */5 * * * *"

543 case "15min":

544 return "3 */15 * * * *"

545 case "30min":

546 return "3 */30 * * * *"

547 case "1hour":

548 return "0 3 * * * *"

549 case "2hour":

550 return "0 3 */2 * * *"

551 case "4hour":

552 return "0 3 */4 * * *"

553 case "6hour":

554 return "0 3 */6 * * *"

555 case "12hour":

556 return "0 3 */12 * * *"

557 case "daily":

558 return "0 3 0 * * *"

559 case _:

560 return None

561 case _:

562 match self.params.get("interval"):

563 case "1m":

564 return "1 * * * * *"

565 case "3m":

566 return "1 */3 * * * *"

567 case "5m":

568 return "1 */5 * * * *"

569 case "10m":

570 return "1 */10 * * * *"

571 case "15m":

572 return "1 */15 * * * *"

573 case "30m":

574 return "1 */30 * * * *"

575 case "1h":

576 return "1 0 * * * *"

577 case "2h":

578 return "1 0 */2 * * *"

579 case "4h":

580 return "1 0 */4 * * *"

581 case "6h":

582 return "1 0 */6 * * *"

583 case "12h":

584 return "1 */12 * * * *"

585 case "1d":

586 return "1 0 0 * * *"

587 case "3d":

588 return "1 0 * */3 * *"

589 case "1w":

590 return "1 0 0 * * 1"

591 case "1M":

592 return "1 0 0 1 * *"

593 case _:

594 return None

595

596 def delay_ms(self) -> int:

597 _SECOND = 1_000

598 _MINUTE = 60 * _SECOND

599 _HOUR = 60 * _MINUTE

600

601 if self.provider == "cryptoquant" and self.endpoint.startswith(

602 "eth/exchange-flows"

603 ):

604 return 3 * _MINUTE

605

606 match self.provider:

607 case "cryptoquant":

608 match self.params.get("window"):

609 case "min":

610 return 10 * _SECOND

611 case "block" | "10min":

612 return 30 * _SECOND

613 case _:

614 return _MINUTE

615 case "glassnode":

616 match self.params.get("i"):

617 case "10m":

618 return 30 * _SECOND

619 case "1w" | "1month":

620 return _HOUR

621 case _:

622 return _MINUTE

623 case "amberdata":

624 match self.params.get("timeInterval"):

625 case "minute":

626 return 10 * _SECOND

627 case "day":

628 return _HOUR

629 case _:

630 return _MINUTE

631 case "coinalyze":

632 match self.params.get("interval"):

633 case "1min" | "5min" | "15min" | "30min":

634 return 10 * _SECOND

635 case "1hour" | "2hour" | "4hour" | "6hour" | "12hour":

636 return _MINUTE

637 case "daily":

638 return _HOUR

639 case _:

640 return _MINUTE

641 case "coinglass" | "coinbase":

642 match self.params.get("interval"):

643 case "1m" | "3m" | "5m":

644 return 10 * _SECOND

645 case "10m" | "15m" | "30m":

646 return 30 * _SECOND

647 case "1h" | "2h" | "4h" | "6h" | "12h":

648 return _MINUTE

649 case "1d" | "3d" | "1w" | "1M":

650 return _HOUR

651 case _:

652 return _MINUTE

653 case _:

654 return _SECOND

655

656 def is_block(self) -> bool:

657 return self.provider == "cryptoquant" and self.params.get("window") == "block"

658

659 def last_closed_time_relative(

660 self, timestamp: datetime, is_collect: bool

661 ) -> datetime | None:

662 interval = self.interval()

663 if interval is None:

664 return None

665

666 interval_ms = int(interval.total_seconds() * 1000)

667 ts_ms = int(timestamp.timestamp() * 1000)

668 truncated_ms = (ts_ms // interval_ms) * interval_ms

669

670 difference_ms = (

671 2 * interval_ms

672 if self.provider == "glassnode"

673 and is_collect

674 and interval_ms == 60 * 60 * 1000

675 else interval_ms

676 )

677

678 return datetime.fromtimestamp(

679 (truncated_ms - difference_ms) / 1000, tz=timezone.utc

680 )

681

682 def last_closed_time(self, is_collect: bool) -> datetime | None:

683 return self.last_closed_time_relative(datetime.now(tz=timezone.utc), is_collect)

684

685

686class Performance(BaseModel):

687 sharpe_ratio: float

688 calmar_ratio: float

689 sortino_ratio: float

690 cagr: float

691 annualized_return: float

692 total_return: float

693 min_cumu: float

694 largest_win: float

695 largest_loss: float

696 num_datapoints: int

697 num_trades: int

698 avg_holding_time_in_seconds: float

699 max_holding_time_in_seconds: float

700 long_trades: int

701 short_trades: int

702 win_trades: int

703 lose_trades: int

704 win_streak: int

705 lose_streak: int

706 win_rate: float

707 start_time: datetime

708 end_time: datetime

709 max_drawdown: float

710 max_drawdown_percentage: float

711 max_drawdown_start_date: datetime

712 max_drawdown_end_date: datetime

713 max_drawdown_recover_date: datetime

714 max_drawdown_max_duration_in_days: float

715 resolution_detected: timedelta | None = None

716 offset_detected: timedelta | None = None

717 datapoints_per_year: float | None = None

718 metadata: dict[str, Any]

719

720 model_config = ConfigDict(extra="allow")

721

722 def __eq__(self, other: object) -> bool:

723 if not isinstance(other, Performance):

724 return False

725 return self.model_dump(exclude={"metadata"}) == other.model_dump(

726 exclude={"metadata"}

727 )

728

729 @staticmethod

730 def from_df(

731 df: pl.DataFrame,

732 start_time: datetime,

733 end_time: datetime,

734 metadata: dict[str, Any] = {},

735 interval: timedelta | None = None,

736 ) -> Self:

737 from adrs.performance.metric import Ratio, Drawdown, Trade

738

739 return Performance.model_validate(

740 {

741 **Ratio(interval=interval).compute(df),

742 **Drawdown(interval=interval).compute(df),

743 **Trade().compute(df),

744 "start_time": start_time,

745 "end_time": end_time,

746 "resolution_detected": interval,

747 "metadata": metadata,

748 }

749 )

750

751

752__all__ = ["Topic", "TopicError", "Symbol", "Data"]