zwitschi
1df4b11aef
CI / lint-test-build (push) Failing after 1m7s
- Introduced new HTML templates for the dashboard, metrics, overview, and backtesting functionalities. - Implemented partial templates for metrics, overview, audit, controls, and charts to enhance modularity. - Updated the Jinja2 template resolution logic to support different deployment environments. - Added a health check template to display the service status. - Included a test suite to verify the template resolution logic. - Updated `pyproject.toml` to include new HTML templates in the package data.
91 lines
3.1 KiB
Python
91 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class TriangularCycle:
|
|
currencies: tuple[str, str, str]
|
|
pairs: tuple[str, str, str]
|
|
|
|
|
|
def _canonical_pair(base: str, quote: str) -> str:
|
|
return f"{base}/{quote}"
|
|
|
|
|
|
class CurrencyGraph:
|
|
def __init__(self) -> None:
|
|
self._adjacency: dict[str, set[str]] = {}
|
|
self._pair_by_direction: dict[tuple[str, str], str] = {}
|
|
|
|
@property
|
|
def adjacency(self) -> dict[str, set[str]]:
|
|
return self._adjacency
|
|
|
|
@property
|
|
def pair_by_direction(self) -> dict[tuple[str, str], str]:
|
|
return self._pair_by_direction
|
|
|
|
def add_pair(self, base: str, quote: str, pair_symbol: str | None = None) -> None:
|
|
normalized_base = base.upper()
|
|
normalized_quote = quote.upper()
|
|
symbol = pair_symbol or _canonical_pair(normalized_base, normalized_quote)
|
|
|
|
self._adjacency.setdefault(normalized_base, set()).add(normalized_quote)
|
|
self._adjacency.setdefault(normalized_quote, set()).add(normalized_base)
|
|
|
|
self._pair_by_direction[(normalized_base, normalized_quote)] = symbol
|
|
self._pair_by_direction[(normalized_quote, normalized_base)] = symbol
|
|
|
|
@classmethod
|
|
def from_kraken_asset_pairs(cls, asset_pairs: dict[str, Any]) -> CurrencyGraph:
|
|
graph = cls()
|
|
for value in asset_pairs.values():
|
|
if not isinstance(value, dict):
|
|
continue
|
|
|
|
wsname = value.get("wsname")
|
|
if isinstance(wsname, str) and "/" in wsname:
|
|
base, quote = wsname.split("/", 1)
|
|
graph.add_pair(base, quote, wsname)
|
|
continue
|
|
|
|
raw_base = value.get("base")
|
|
raw_quote = value.get("quote")
|
|
if isinstance(raw_base, str) and isinstance(raw_quote, str):
|
|
graph.add_pair(raw_base, raw_quote)
|
|
|
|
return graph
|
|
|
|
def triangular_cycles(self) -> list[TriangularCycle]:
|
|
found: dict[tuple[str, str, str], TriangularCycle] = {}
|
|
|
|
for a, neighbors_a in self._adjacency.items():
|
|
for b in neighbors_a:
|
|
if a >= b:
|
|
continue
|
|
neighbors_b = self._adjacency.get(b, set())
|
|
for c in neighbors_b:
|
|
if b >= c:
|
|
continue
|
|
if a not in self._adjacency.get(c, set()):
|
|
continue
|
|
|
|
p_ab = self._pair_by_direction[(a, b)]
|
|
p_bc = self._pair_by_direction[(b, c)]
|
|
p_ca = self._pair_by_direction[(c, a)]
|
|
|
|
key = (a, b, c)
|
|
found[key] = TriangularCycle(currencies=key, pairs=(p_ab, p_bc, p_ca))
|
|
|
|
return list(found.values())
|
|
|
|
@staticmethod
|
|
def index_cycles_by_pair(cycles: list[TriangularCycle]) -> dict[str, list[TriangularCycle]]:
|
|
index: dict[str, list[TriangularCycle]] = {}
|
|
for cycle in cycles:
|
|
for pair in cycle.pairs:
|
|
index.setdefault(pair, []).append(cycle)
|
|
return index
|