Разработка на Python

from dataclasses import dataclass

@dataclass(frozen=True)
class SizeResult:
    qty: float
    risk_cash: float
    clipped: bool

def position_size(
    equity: float,
    risk_pct: float,
    entry: float,
    stop: float,
    point_value: float = 1.0,
    qty_step: float = 0.001,
    max_qty: float | None = None,
) -> SizeResult:
    if equity <= 0:
        raise ValueError("equity must be positive")
    stop_dist = abs(entry - stop)
    if stop_dist <= 0:
        raise ValueError("stop must differ from entry")
    risk_cash = equity * (risk_pct / 100.0)
    raw = risk_cash / (stop_dist * point_value)
    steps = max(1, int(raw / qty_step))
    qty = steps * qty_step
    clipped = False
    if max_qty is not None and qty > max_qty:
        qty, clipped = max_qty, True
    return SizeResult(qty=qty, risk_cash=risk_cash, clipped=clipped)

from datetime import date, timezone, datetime

class KillSwitch:
    def __init__(self, daily_limit_pct: float):
        self.daily_limit_pct = daily_limit_pct
        self._day = date.min
        self._start_equity = 0.0
        self.tripped = False

    def on_mark(self, equity: float, now: datetime | None = None) -> bool:
        now = now or datetime.now(timezone.utc)
        d = now.date()
        if d != self._day:
            self._day = d
            self._start_equity = equity
            self.tripped = False
        if self._start_equity <= 0:
            return False
        dd = (self._start_equity - equity) / self._start_equity * 100
        if dd >= self.daily_limit_pct:
            self.tripped = True
        return self.tripped

import random, time
import requests

def request_with_retry(method: str, url: str, **kwargs) -> requests.Response:
    delays = (0.4, 1.0, 2.5, 5.0)
    last_exc: Exception | None = None
    for i, delay in enumerate(delays, start=1):
        try:
            r = requests.request(method, url, timeout=8, **kwargs)
            if r.status_code in (429, 500, 502, 503, 504):
                raise requests.HTTPError(f"retryable {r.status_code}")
            r.raise_for_status()
            return r
        except (requests.Timeout, requests.ConnectionError, requests.HTTPError) as exc:
            last_exc = exc
            jitter = random.random() * 0.2
            time.sleep(delay + jitter)
    raise RuntimeError("exchange unreachable") from last_exc

def atr(high: list[float], low: list[float], close: list[float], n: int = 14) -> float:
    trs = []
    for i in range(1, len(close)):
        tr = max(
            high[i] - low[i],
            abs(high[i] - close[i - 1]),
            abs(low[i] - close[i - 1]),
        )
        trs.append(tr)
    window = trs[-n:]
    return sum(window) / len(window)

def trail_stop(side: str, price: float, atr_val: float, k: float = 2.0) -> float:
    dist = k * atr_val
    return price - dist if side == "long" else price + dist

import hashlib, json

def client_order_id(signal_id: str, bar_ts: int, side: str) -> str:
    payload = json.dumps({"s": signal_id, "t": bar_ts, "side": side}, sort_keys=True)
    return hashlib.sha1(payload.encode()).hexdigest()[:24]

def already_sent(store: set[str], oid: str) -> bool:
    if oid in store:
        return True
    store.add(oid)
    return False