main.py

"""
Клиент API AML-скоринга Atlorium — скрининг криптокошелька по адресу и сети.

Запуск (работает сразу, без регистрации — на демо-ключе):
    pip install -r requirements.txt
    python main.py
    python main.py 0x26a5919cd0392df875a53873b0323c42094bb216 ETH

Боевой ключ: получить на https://atlorium.com и положить в переменную окружения
ATLORIUM_API_KEY. Код при этом не меняется.
"""

import os
import sys
from dataclasses import dataclass, field

import requests

# Публичный демо-ключ. С ним API отвечает правдоподобными МОКАМИ (не реальными
# данными) — чтобы можно было встроить и протестировать интеграцию до оплаты.
# Ответы детерминированы: один и тот же запрос всегда даёт один и тот же результат,
# поэтому на них можно писать стабильные тесты.
SANDBOX_KEY = "ak_sandbox_demo_mockdata_v1"

API_KEY = os.environ.get("ATLORIUM_API_KEY", SANDBOX_KEY)
BASE_URL = os.environ.get("ATLORIUM_BASE_URL", "https://atlorium.com")

TIMEOUT = 60


class AtloriumError(RuntimeError):
    """Ошибка API. Код HTTP разложен в человекочитаемую причину."""

    REASONS = {
        400: "Адрес или сеть не указаны либо имеют неверный формат (неподдерживаемая сеть, некорректный адрес)",
        401: "API-ключ отсутствует, просрочен или недействителен",
        402: "Недостаточно кредитов на балансе — пополните на https://atlorium.com",
        404: "Объект не найден",
        429: "Превышен лимит запросов — повторите позже",
        503: "Сервис AML-скоринга временно недоступен (за сбой на своей стороне мы не списываем деньги)",
    }

    def __init__(self, status: int, body: str):
        reason = self.REASONS.get(status, "Неизвестная ошибка")
        super().__init__(f"HTTP {status}: {reason}. Ответ сервера: {body[:200]}")
        self.status = status


HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Accept": "application/json",
}


def _post(path: str, payload: dict) -> dict:
    """POST с JSON-телом. Скрининг — это не GET: адрес и сеть уходят в теле запроса."""
    response = requests.post(
        f"{BASE_URL}{path}",
        json=payload,
        headers={**HEADERS, "Content-Type": "application/json"},
        timeout=TIMEOUT,
    )
    if not response.ok:
        raise AtloriumError(response.status_code, response.text)
    return response.json()


def _get(path: str, params: dict) -> dict:
    response = requests.get(f"{BASE_URL}{path}", params=params, headers=HEADERS, timeout=TIMEOUT)
    if not response.ok:
        raise AtloriumError(response.status_code, response.text)
    return response.json()


def screen_wallet(wallet_address: str, blockchain: str) -> dict:
    """AML-скрининг кошелька: балл риска 0–100, санкции и PEP, источники и
    назначение средств, контрагенты.

    Тарифицируется только успешный скрининг. Неверный ввод и недоступность
    источника деньги не списывают.
    """
    return _post("/api/aml/screen", {"walletAddress": wallet_address, "blockchain": blockchain})


def validate_wallet(wallet_address: str, blockchain: str) -> dict:
    """Проверка ФОРМАТА адреса и поддержки сети — без обращения к блокчейну.

    Отвечает на вопрос «можно ли вообще скринить такую пару адрес+сеть»,
    а не «существует ли этот адрес». Учитывается так же, как и скрининг:
    это обращение к сервису.
    """
    return _get("/api/aml/validate", {"walletAddress": wallet_address, "blockchain": blockchain})


# ── Применение данных: комплаенс-гейт перед выводом или зачислением средств ────
# Карточка скрининга сама по себе — просто JSON. Ценность появляется, когда по ней
# принимают решение. Ниже — гейт из трёх исходов: ALLOW / REVIEW / BLOCK.

# ВНИМАНИЕ: пороги ниже — отправная точка, а не универсальная истина.
# Настраивайте их под свою риск-политику, юрисдикцию и аппетит к риску.
SCORE_BLOCK = 75  # балл >= этого — блокируем
SCORE_REVIEW = 40  # балл >= этого — на ручную проверку
EXPOSURE_ALERT_PCT = 5.0  # доля средств из «плохой» категории, с которой это уже сигнал

# Категории источников/назначения средств, которые сами по себе поднимают флаг.
# Сравнение по подстроке — написание кодов категорий может различаться.
HIGH_RISK_CATEGORY_KEYWORDS = (
    "sanction", "mixer", "tumbler", "darknet", "ransomware",
    "stolen", "scam", "fraud", "terror", "illicit", "hack", "phishing",
)

ALLOW, REVIEW, BLOCK = "ALLOW", "REVIEW", "BLOCK"
_RANK = {ALLOW: 0, REVIEW: 1, BLOCK: 2}


@dataclass
class GateResult:
    """Решение комплаенс-гейта и факторы, которые на него повлияли."""

    decision: str = ALLOW
    factors: list[str] = field(default_factory=list)

    def escalate(self, decision: str) -> None:
        """Решение может только ужесточаться: ALLOW → REVIEW → BLOCK."""
        if _RANK[decision] > _RANK[self.decision]:
            self.decision = decision


def _is_high_risk_category(category: str | None) -> bool:
    lowered = (category or "").lower()
    return any(keyword in lowered for keyword in HIGH_RISK_CATEGORY_KEYWORDS)


def compliance_gate(result: dict) -> GateResult:
    """Комплаенс-гейт: пропустить операцию, отправить на ручную проверку или заблокировать."""
    gate = GateResult()

    # 1. Санкции — безусловный стоп. Никакой низкий балл это не перевешивает.
    if result.get("sanctionsHit"):
        gate.escalate(BLOCK)
        gate.factors.append("[!] Санкционное совпадение — блокировка безусловна")
    else:
        gate.factors.append("[+] Санкционных совпадений нет")

    # 2. Публичное должностное лицо (PEP) среди контрагентов — тоже безусловный стоп:
    #    такие операции требуют отдельного решения комплаенс-офицера, а не автомата.
    if result.get("pepCounterparty"):
        gate.escalate(BLOCK)
        gate.factors.append("[!] Контрагент из списка PEP — блокировка безусловна")
    else:
        gate.factors.append("[+] PEP-контрагентов нет")

    # 3. Балл риска 0–100.
    score = int(result.get("riskScore") or 0)
    severity = result.get("severity") or "?"
    if score >= SCORE_BLOCK:
        gate.escalate(BLOCK)
        gate.factors.append(f"[!] Риск-балл {score} >= порога блокировки {SCORE_BLOCK} (severity {severity})")
    elif score >= SCORE_REVIEW:
        gate.escalate(REVIEW)
        gate.factors.append(f"[?] Риск-балл {score} в диапазоне ручной проверки {SCORE_REVIEW}–{SCORE_BLOCK - 1}")
    else:
        gate.factors.append(f"[+] Риск-балл {score} ниже порога ручной проверки ({SCORE_REVIEW})")

    # 4. Экспозиция на токсичные категории: миксеры, даркнет, вымогатели, краденое.
    #    Балл может быть умеренным, а вот прямая связь с миксером — нет.
    exposures = (result.get("sourceOfFunds") or []) + (result.get("destinationOfFunds") or [])
    for exposure in exposures:
        percentage = float(exposure.get("percentage") or 0)
        if _is_high_risk_category(exposure.get("category")) and percentage >= EXPOSURE_ALERT_PCT:
            gate.escalate(REVIEW)
            gate.factors.append(
                f"[?] Экспозиция {percentage:.1f}% на категорию «{exposure.get('category')}»"
                f" ({exposure.get('entityName') or 'контрагент не назван'})"
            )

    if result.get("dominantRiskCategory"):
        gate.factors.append(f"[i] Доминирующая категория риска: {result['dominantRiskCategory']}")

    return gate


# ── Печать отчёта ─────────────────────────────────────────────────────────────


def _money(value) -> str:
    return "н/д" if value is None else f"${float(value):,.2f}".replace(",", " ")


def _print_exposures(title: str, exposures: list[dict]) -> None:
    if not exposures:
        return
    print(f"  {title}:")
    for exposure in exposures:
        parts = [
            f"{float(exposure.get('percentage') or 0):.1f}%",
            str(exposure.get("category") or "—"),
            str(exposure.get("entityName") or "—"),
        ]
        if exposure.get("description"):
            parts.append(str(exposure["description"]))
        if exposure.get("country"):
            parts.append(str(exposure["country"]))
        print("    " + " · ".join(parts))


def main() -> int:
    if API_KEY == SANDBOX_KEY:
        print("Демо-ключ: ответы сгенерированы (моки), не реальные данные.\n")

    wallet = sys.argv[1] if len(sys.argv) > 1 else "1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa"
    blockchain = sys.argv[2] if len(sys.argv) > 2 else "BTC"

    try:
        result = screen_wallet(wallet, blockchain)
    except AtloriumError as error:
        print(f"Ошибка: {error}", file=sys.stderr)
        return 1

    print(f"{result['blockchain']} {result['walletAddress']}")
    print(f"  Риск-балл: {result['riskScore']}/100 · severity: {result['severity']} · статус: {result['status']}")
    if result.get("summary"):
        print(f"  Резюме: {result['summary']}")
    print(f"  Санкции: {'ДА' if result.get('sanctionsHit') else 'нет'}"
          f" · PEP: {'ДА' if result.get('pepCounterparty') else 'нет'}")
    print(f"  Оборот: {_money(result.get('totalValueUsd'))} · комиссии: {_money(result.get('feeUsd'))}")

    _print_exposures("Источники средств", result.get("sourceOfFunds") or [])
    _print_exposures("Назначение средств", result.get("destinationOfFunds") or [])

    counterparties = result.get("counterpartyConnections") or []
    if counterparties:
        print("  Контрагенты:")
        for counterparty in counterparties:
            print(
                f"    {counterparty.get('entityName') or '—'}"
                f" · {counterparty.get('entityType') or '—'}"
                f" · риск {counterparty.get('riskLevel') or '—'}"
                f" · {float(counterparty.get('percentage') or 0):.1f}%"
            )

    gate = compliance_gate(result)
    print(f"\nКОМПЛАЕНС-ГЕЙТ: {gate.decision}")
    for factor in gate.factors:
        print(f"  {factor}")

    return 0


if __name__ == "__main__":
    raise SystemExit(main())