#!/usr/bin/env python3 """ Продвинутый анализатор транзакций с отслеживанием жизненного цикла ордеров Учитывает не только изменения объемов, но и появление/исчезновение ордеров """ import asyncio import time import hashlib from typing import Dict, List, Optional, Set, Tuple from datetime import datetime, timedelta from decimal import Decimal from dataclasses import dataclass from loguru import logger from ..models.order import P2POrder, OrderBookSnapshot from ..models.transaction import ( DVariable, MerchantLiquidityStats, TransactionType, IndividualTransaction ) #TODO: переделать на поиск по объему, а не по цене @dataclass class MerchantOrderState: """Состояние ордеров мерчанта""" merchant_id: str merchant_name: str asset: str fiat: str # Текущие активные ордера active_orders: Dict[str, P2POrder] # order_id -> order # История изменений last_seen: datetime total_volume_history: List[Tuple[datetime, float]] # (timestamp, total_volume) def get_total_volume(self) -> float: """Получить общий объем всех активных ордеров мерчанта""" return sum(order.available_amount for order in self.active_orders.values()) def update_volume_history(self): """Обновить историю объемов""" current_volume = self.get_total_volume() self.total_volume_history.append((datetime.now(), current_volume)) # Оставляем только последние 10 записей if len(self.total_volume_history) > 10: self.total_volume_history = self.total_volume_history[-10:] @dataclass class OrderTransition: """Переход состояния ордеров мерчанта""" merchant_id: str merchant_name: str asset: str fiat: str trade_type: str timestamp: datetime # Состояние до orders_before: Dict[str, P2POrder] total_volume_before: float # Состояние после orders_after: Dict[str, P2POrder] total_volume_after: float # Классификация disappeared_orders: List[P2POrder] appeared_orders: List[P2POrder] modified_orders: List[Tuple[P2POrder, P2POrder]] # (old, new) volume_change: float = 0.0 def __post_init__(self): self.volume_change = self.total_volume_after - self.total_volume_before class AdvancedTransactionAnalyzer: """ Продвинутый анализатор P2P транзакций с отслеживанием жизненного цикла ордеров """ def __init__(self, d_variable_timeout: int = 3600): # 60 минут по умолчанию self.d_variable_timeout = d_variable_timeout # Отслеживание состояния мерчантов self.merchant_states: Dict[str, MerchantOrderState] = {} # merchant_key -> state # Классические D-переменные для совместимости self.d_variables: Dict[str, DVariable] = {} # Статистика self.merchant_stats: Dict[str, MerchantLiquidityStats] = {} # Дедупликация self.processed_transactions: Set[str] = set() # Последние снапшоты для отслеживания изменений self.last_snapshots: Dict[str, OrderBookSnapshot] = {} logger.info("🚀 Advanced Transaction Analyzer initialized") def _get_merchant_key(self, merchant_id: str, asset: str, fiat: str) -> str: """Получить ключ для идентификации мерчанта""" return f"{merchant_id}_{asset}_{fiat}" def _initialize_merchant_state(self, merchant: P2POrder) -> MerchantOrderState: """Инициализировать состояние мерчанта""" return MerchantOrderState( merchant_id=merchant.merchant_id, merchant_name=merchant.merchant_name, asset=merchant.asset, fiat=merchant.fiat, active_orders={merchant.order_id: merchant}, last_seen=datetime.now(), total_volume_history=[] ) def analyze_orderbook_changes( self, current_snapshot: OrderBookSnapshot, previous_snapshot: Optional[OrderBookSnapshot] = None ) -> List: """Анализ изменений в ордербуке с отслеживанием жизненного цикла ордеров""" events = [] snapshot_key = f"{current_snapshot.asset}_{current_snapshot.trade_type}_{current_snapshot.fiat}" # Используем предыдущий snapshot из кэша, если не передан if previous_snapshot is None: previous_snapshot = self.last_snapshots.get(snapshot_key) # Сохраняем текущий snapshot self.last_snapshots[snapshot_key] = current_snapshot if previous_snapshot is None: # Первый снапшот - инициализируем состояния мерчантов self._initialize_merchant_states_from_snapshot(current_snapshot) logger.info(f"🔍 First snapshot for {snapshot_key}, initialized {len(current_snapshot.orders)} merchants") return events # Анализируем переходы состояний мерчантов transitions = self._analyze_merchant_transitions(current_snapshot, previous_snapshot) # Обрабатываем каждый переход for transition in transitions: transition_events = self._process_merchant_transition(transition) events.extend(transition_events) # Обновляем состояния мерчантов self._update_merchant_states_from_snapshot(current_snapshot) # Очистка истёкших D-переменных self._cleanup_expired_d_variables() return events def _initialize_merchant_states_from_snapshot(self, snapshot: OrderBookSnapshot): """Инициализировать состояния мерчантов из снапшота""" for order in snapshot.orders: merchant_key = self._get_merchant_key(order.merchant_id, order.asset, order.fiat) if merchant_key not in self.merchant_states: self.merchant_states[merchant_key] = self._initialize_merchant_state(order) else: # Добавляем ордер к существующему мерчанту self.merchant_states[merchant_key].active_orders[order.order_id] = order #TODO: переделать на поиск по объему, а не по ценеs def _update_merchant_states_from_snapshot(self, snapshot: OrderBookSnapshot): """Обновить состояния мерчантов из снапшота""" # Сначала собираем все ордера по мерчантам из текущего снапшота current_merchants = {} for order in snapshot.orders: merchant_key = self._get_merchant_key(order.merchant_id, order.asset, order.fiat) if merchant_key not in current_merchants: current_merchants[merchant_key] = {} current_merchants[merchant_key][order.order_id] = order # Обновляем состояния for merchant_key, orders in current_merchants.items(): if merchant_key in self.merchant_states: self.merchant_states[merchant_key].active_orders = orders self.merchant_states[merchant_key].last_seen = datetime.now() self.merchant_states[merchant_key].update_volume_history() else: # Новый мерчант first_order = next(iter(orders.values())) self.merchant_states[merchant_key] = MerchantOrderState( merchant_id=first_order.merchant_id, merchant_name=first_order.merchant_name, asset=first_order.asset, fiat=first_order.fiat, active_orders=orders, last_seen=datetime.now(), total_volume_history=[] ) self.merchant_states[merchant_key].update_volume_history() # Помечаем исчезнувших мерчантов for merchant_key in list(self.merchant_states.keys()): if merchant_key not in current_merchants: self.merchant_states[merchant_key].active_orders = {} self.merchant_states[merchant_key].update_volume_history() def _analyze_merchant_transitions( self, current_snapshot: OrderBookSnapshot, previous_snapshot: OrderBookSnapshot ) -> List[OrderTransition]: """Анализировать переходы состояний мерчантов между снапшотами""" # Группируем ордера по мерчантам def group_by_merchants(snapshot): groups = {} for order in snapshot.orders: merchant_key = self._get_merchant_key(order.merchant_id, order.asset, order.fiat) if merchant_key not in groups: groups[merchant_key] = {} groups[merchant_key][order.order_id] = order return groups previous_merchants = group_by_merchants(previous_snapshot) current_merchants = group_by_merchants(current_snapshot) transitions = [] # Анализируем всех мерчантов (и исчезнувших, и появившихся) all_merchant_keys = set(previous_merchants.keys()) | set(current_merchants.keys()) for merchant_key in all_merchant_keys: orders_before = previous_merchants.get(merchant_key, {}) orders_after = current_merchants.get(merchant_key, {}) # Пропускаем, если нет изменений if orders_before == orders_after: continue # Вычисляем изменения total_volume_before = sum(order.available_amount for order in orders_before.values()) total_volume_after = sum(order.available_amount for order in orders_after.values()) disappeared_orders = [order for order_id, order in orders_before.items() if order_id not in orders_after] appeared_orders = [order for order_id, order in orders_after.items() if order_id not in orders_before] modified_orders = [] for order_id in set(orders_before.keys()) & set(orders_after.keys()): old_order = orders_before[order_id] new_order = orders_after[order_id] if old_order.available_amount != new_order.available_amount: modified_orders.append((old_order, new_order)) # ФИЛЬТРАЦИЯ МАНИПУЛЯЦИЙ: проверяем на подозрительные паттерны if self._is_likely_manipulation(total_volume_before, total_volume_after, disappeared_orders, appeared_orders, merchant_key): logger.debug(f"🎭 Filtering manipulation: {merchant_key} - " f"volume {total_volume_before:.2f} → {total_volume_after:.2f}") continue # Создаем переход только если есть значимые изменения if (abs(total_volume_after - total_volume_before) > 0.01 or disappeared_orders or appeared_orders or modified_orders): # Получаем информацию о мерчанте merchant_info = (list(orders_after.values()) or list(orders_before.values()))[0] transition = OrderTransition( merchant_id=merchant_info.merchant_id, merchant_name=merchant_info.merchant_name, asset=merchant_info.asset, fiat=merchant_info.fiat, trade_type=current_snapshot.trade_type, timestamp=datetime.now(), orders_before=orders_before, total_volume_before=total_volume_before, orders_after=orders_after, total_volume_after=total_volume_after, disappeared_orders=disappeared_orders, appeared_orders=appeared_orders, modified_orders=modified_orders ) transitions.append(transition) return transitions def _is_likely_manipulation(self, volume_before: float, volume_after: float, disappeared_orders: List, appeared_orders: List, merchant_key: str) -> bool: """Проверить, является ли изменение манипуляцией мерчанта""" # Получаем историю этого мерчанта if merchant_key not in self.merchant_states: return False merchant_state = self.merchant_states[merchant_key] # Проверка 1: Круговые изменения (A→B→A) if len(merchant_state.total_volume_history) >= 3: recent_volumes = [vol for _, vol in merchant_state.total_volume_history[-3:]] current_volume = volume_after # Если текущий объем равен объему 2 шага назад if len(recent_volumes) >= 2 and abs(current_volume - recent_volumes[0]) < 0.01: logger.debug(f"🔄 Circular pattern detected for {merchant_key}: " f"{recent_volumes[0]:.2f} → {recent_volumes[1]:.2f} → {current_volume:.2f}") return True # Убираем фильтр по большим объемам: для USDT/USDC типичны десятки тысяч # max_volume = max(volume_before, volume_after) # if max_volume > 1000: # return True # Проверка 3: Полное исчезновение и появление одновременно if (volume_before > 0 and volume_after > 0 and len(disappeared_orders) > 0 and len(appeared_orders) > 0 and abs(volume_before - volume_after) < 0.01): logger.debug(f"🔀 Order reshuffling detected for {merchant_key}") return True return False def _process_merchant_transition(self, transition: OrderTransition) -> List: """Обработать переход состояния мерчанта и классифицировать как сделку/пополнение/отмену""" events = [] # Логирование для отладки logger.debug(f"🔄 Processing transition for {transition.merchant_name}: " f"{transition.total_volume_before:.2f} → {transition.total_volume_after:.2f} " f"({transition.volume_change:+.2f}) {transition.asset}") # Классифицируем тип изменения if abs(transition.volume_change) < 0.01: # Объем не изменился, но ордера поменялись - возможно, изменение цены self._process_neutral_transition(transition) elif transition.volume_change < -0.01: # Объем уменьшился - возможна реальная сделка trade_event = self._process_volume_decrease_transition(transition) if trade_event: events.append(trade_event) elif transition.volume_change > 0.01: # Объем увеличился - пополнение или подтверждение сделки increase_event = self._process_volume_increase_transition(transition) if increase_event: events.append(increase_event) return events #TODO: переделать на поиск по объему, а не по цене def _process_volume_decrease_transition(self, transition: OrderTransition): """Обработать уменьшение объема (создание D-переменной для потенциальной сделки)""" decrease_amount = abs(transition.volume_change) # Создаем D-переменную для отслеживания потенциальной сделки transaction_hash = self._create_transaction_hash( transition.merchant_id, transition.asset, decrease_amount, transition.total_volume_before, transition.total_volume_after, "volume_decrease" ) # Проверяем дедупликацию if transaction_hash in self.processed_transactions: logger.debug(f"🔄 Duplicate volume decrease detected for {transition.merchant_name}, skipping") return None self.processed_transactions.add(transaction_hash) # Создаем D-переменную d_var_id = f"{transition.merchant_id}_{transition.asset}_{transaction_hash[:8]}" d_variable = DVariable( id=d_var_id, merchant_id=transition.merchant_id, asset=transition.asset, fiat=transition.fiat, trade_type=transition.trade_type, amount=decrease_amount, timestamp=datetime.now(), expires_at=datetime.now() + timedelta(seconds=self.d_variable_timeout), price_at_d1=self._estimate_price_from_transition(transition), merchant_name=transition.merchant_name ) self.d_variables[d_var_id] = d_variable logger.debug(f"📉 Created D-variable: {d_var_id} for {decrease_amount} {transition.asset} " f"(expires in {self.d_variable_timeout}s)") # НЕ создаем транзакцию сразу - ждем подтверждения через увеличение объема # или истечение D-переменной return None # Не возвращаем событие, пока не подтвердится def _process_volume_increase_transition(self, transition: OrderTransition): """Обработать увеличение объема (поиск соответствующей D-переменной или пополнение)""" increase_amount = transition.volume_change # Создаем хеш для дедупликации transaction_hash = self._create_transaction_hash( transition.merchant_id, transition.asset, increase_amount, transition.total_volume_before, transition.total_volume_after, "volume_increase" ) if transaction_hash in self.processed_transactions: logger.debug(f"🔄 Duplicate volume increase detected for {transition.merchant_name}, skipping") return None self.processed_transactions.add(transaction_hash) # Ищем соответствующую D-переменную (отмена заявки) с USD-нормализацией допуска matching_d_var = None for d_var_id, d_var in self.d_variables.items(): if d_var.merchant_id != transition.merchant_id or d_var.asset != transition.asset: continue try: diff = abs(float(d_var.amount) - float(increase_amount)) # оценим USD-эквивалент по цене из перехода или D1 price = self._estimate_price_from_transition(transition) or d_var.price_at_d1 or 0.0 diff_usd = diff * price if price > 0 else diff base_usd_tol = max(5.0, 0.02 * abs(float(d_var.amount) * price) if price > 0 else 0.02 * abs(float(d_var.amount))) # если цены нет, используем 2% от объема или 0.02 if price <= 0: tolerance = max(0.02, 0.02 * max(float(d_var.amount), float(increase_amount))) cond = diff <= tolerance else: cond = diff_usd <= base_usd_tol except Exception: cond = False if cond: matching_d_var = d_var break if matching_d_var: # Найдена соответствующая D-переменная - это отмена заявки logger.info(f"🚫 Order cancellation detected: {transition.merchant_name} " f"{increase_amount:.2f} {transition.asset} " f"(time since decrease: {(datetime.now() - matching_d_var.timestamp).total_seconds():.0f}s)") # Удаляем D-переменную del self.d_variables[matching_d_var.id] # Создаем транзакцию отмены заявки transaction = IndividualTransaction( merchant_id=transition.merchant_id, merchant_name=transition.merchant_name, asset=transition.asset, fiat=transition.fiat, transaction_type="order_cancellation", amount=increase_amount, price=self._estimate_price_from_transition(transition), volume_before=transition.total_volume_before, volume_after=transition.total_volume_after, trade_type=transition.trade_type ) # Обновляем статистику мерчанта self._update_merchant_stats_with_transaction( transition.merchant_id, transition.merchant_name, transition.asset, "order_cancellation", transaction ) return transaction else: # Не найдена соответствующая D-переменная - это пополнение мерчанта logger.info(f"💰 Merchant top-up detected: {transition.merchant_name} " f"+{increase_amount:.2f} {transition.asset} " f"(volume: {transition.total_volume_before:.2f} → {transition.total_volume_after:.2f})") # Создаем транзакцию пополнения transaction = IndividualTransaction( merchant_id=transition.merchant_id, merchant_name=transition.merchant_name, asset=transition.asset, fiat=transition.fiat, transaction_type="top_up", amount=increase_amount, price=self._estimate_price_from_transition(transition), volume_before=transition.total_volume_before, volume_after=transition.total_volume_after, trade_type=transition.trade_type ) # Обновляем статистику мерчанта self._update_merchant_stats_with_transaction( transition.merchant_id, transition.merchant_name, transition.asset, "top_up", transaction ) return transaction def _process_neutral_transition(self, transition: OrderTransition): """Обработать нейтральный переход (объем не изменился)""" # Логируем для отладки, но не создаем транзакцию if transition.disappeared_orders or transition.appeared_orders: logger.debug(f"🔄 Order refresh: {transition.merchant_name} " f"(disappeared: {len(transition.disappeared_orders)}, " f"appeared: {len(transition.appeared_orders)})") def _estimate_price_from_transition(self, transition: OrderTransition) -> float: """Оценить цену из перехода состояния""" # Берем среднюю цену из всех ордеров после перехода if transition.orders_after: prices = [order.price for order in transition.orders_after.values()] return sum(prices) / len(prices) elif transition.orders_before: prices = [order.price for order in transition.orders_before.values()] return sum(prices) / len(prices) else: return 0.0 def _update_merchant_stats_with_transaction(self, merchant_id: str, merchant_name: str, asset: str, event_type: str, transaction: IndividualTransaction): """Обновление статистики мерчанта с индивидуальной транзакцией""" if merchant_id not in self.merchant_stats: self.merchant_stats[merchant_id] = MerchantLiquidityStats( merchant_id=merchant_id, merchant_name=merchant_name, asset=asset ) stats = self.merchant_stats[merchant_id] if not stats.merchant_name and merchant_name: stats.merchant_name = merchant_name if event_type == "real_trade": stats.add_confirmed_trade(transaction) elif event_type == "order_cancellation": stats.add_cancelled_order(transaction) elif event_type == "top_up": stats.add_merchant_topup(transaction) def _create_transaction_hash(self, merchant_id: str, asset: str, amount: float, volume_before: float, volume_after: float, transaction_type: str) -> str: """Создание уникального хеша для дедупликации транзакций""" hash_string = f"{merchant_id}_{asset}_{amount:.6f}_{volume_before:.6f}_{volume_after:.6f}_{transaction_type}" return hashlib.sha256(hash_string.encode()).hexdigest()[:16] def _cleanup_expired_d_variables(self): """Эскалация подтверждений D‑переменных и финальное подтверждение реальных сделок""" current_time = datetime.now() to_delete = [] # Временные ступени подтверждения: 10m, 20m, 30m, 60m stages_seconds = [600, 1200, 1800, 3600] for d_var_id, d_var in self.d_variables.items(): age_seconds = (current_time - d_var.timestamp).total_seconds() # Эскалация ступени next_stage = d_var.confirmation_stage while next_stage < len(stages_seconds) and age_seconds >= stages_seconds[next_stage]: next_stage += 1 if next_stage != d_var.confirmation_stage: d_var.confirmation_stage = next_stage d_var.last_confirmation_at = current_time logger.info( f"⏱️ D-stage {d_var.confirmation_stage}/4 for {d_var.merchant_id} " f"-{float(d_var.amount):.2f} {d_var.asset} (age {int(age_seconds)}s)" ) # Финальное подтверждение сделки строго по достижению 4 стадии if d_var.is_matched: to_delete.append(d_var_id) continue if d_var.confirmation_stage >= 4: logger.info( f"✅ Real trade auto-confirmed: merchant {d_var.merchant_id} " f"-{float(d_var.amount):.2f} {d_var.asset} (age {int(age_seconds)}s)" ) # Найдём имя/параметры из текущего состояния, если доступны state_key = self._get_merchant_key(d_var.merchant_id, d_var.asset, d_var.fiat) state = self.merchant_states.get(state_key) resolved_name = d_var.merchant_name or (state.merchant_name if state else "") resolved_trade_type = d_var.trade_type or (state and getattr(state, 'trade_type', None)) or "SELL" resolved_fiat = d_var.fiat resolved_price = d_var.price_at_d1 or (self._estimate_price_from_transition( OrderTransition( merchant_id=d_var.merchant_id, merchant_name=resolved_name, asset=d_var.asset, fiat=resolved_fiat, timestamp=current_time, orders_before=state.active_orders if state else {}, total_volume_before=state.get_total_volume() if state else 0.0, orders_after=state.active_orders if state else {}, total_volume_after=state.get_total_volume() if state else 0.0, disappeared_orders=[], appeared_orders=[], modified_orders=[], trade_type=resolved_trade_type ) ) if state else 0.0) transaction = IndividualTransaction( merchant_id=d_var.merchant_id, merchant_name=resolved_name, asset=d_var.asset, fiat=resolved_fiat, transaction_type="real_trade", amount=float(d_var.amount), price=resolved_price, volume_before=0.0, volume_after=0.0, trade_type=resolved_trade_type ) self._update_merchant_stats_with_transaction( d_var.merchant_id, resolved_name, d_var.asset, "real_trade", transaction ) to_delete.append(d_var_id) for d_var_id in to_delete: logger.debug(f"🗑️ Removing D-variable: {d_var_id}") del self.d_variables[d_var_id] def export_stats_to_json(self) -> dict: """Экспорт статистики в JSON формат""" def serialize_transaction(transaction: IndividualTransaction) -> dict: return transaction.to_dict() # Aggregate pending (unconfirmed) real trades per merchant from active D-variables pending_by_merchant: Dict[str, Dict[str, float]] = {} for d_var in self.d_variables.values(): merchant_id = d_var.merchant_id if merchant_id not in pending_by_merchant: pending_by_merchant[merchant_id] = {"count": 0, "volume": 0.0} pending_by_merchant[merchant_id]["count"] += 1 # Safe float conversion for Decimal/float try: pending_by_merchant[merchant_id]["volume"] += float(d_var.amount) except Exception: pending_by_merchant[merchant_id]["volume"] += 0.0 merchant_stats_json = {} for merchant_id, stats in self.merchant_stats.items(): merchant_stats_json[merchant_id] = { "merchant_name": stats.merchant_name, "asset": stats.asset, "real_trades_count": stats.real_trades_count, "real_trades_volume": stats.real_trades_volume, "order_cancellations_count": stats.order_cancellations_count, "order_cancellations_volume": stats.order_cancellations_volume, "top_ups_count": stats.top_ups_count, "top_ups_volume": stats.top_ups_volume, "last_activity": stats.last_activity.isoformat() if stats.last_activity else None, # Pending (unconfirmed) trades derived from active D-variables "pending_real_trades_count": int(pending_by_merchant.get(merchant_id, {}).get("count", 0)), "pending_real_trades_volume": float(pending_by_merchant.get(merchant_id, {}).get("volume", 0.0)), "individual_trades": [serialize_transaction(t) for t in stats.individual_trades], "individual_top_ups": [serialize_transaction(t) for t in stats.individual_top_ups], "individual_cancellations": [serialize_transaction(t) for t in stats.individual_cancellations], } # Дополнительно наполним статистику базовыми данными из текущих состояний мерчантов, # чтобы дашборд отображал активных мерчантов даже при отсутствии событий. for state_key, state in self.merchant_states.items(): # Сформируем уникальный ключ merchant_id+asset+fiat, чтобы не смешивать активы derived_key = f"{state.merchant_id}_{state.asset}_{state.fiat}" # Не дублируем, если по мерчанту уже есть агрегированная статистика по событиям if state.merchant_id in merchant_stats_json or derived_key in merchant_stats_json: continue merchant_stats_json[derived_key] = { "merchant_name": state.merchant_name, "asset": state.asset, "real_trades_count": 0, "real_trades_volume": 0.0, "order_cancellations_count": 0, "order_cancellations_volume": 0.0, "top_ups_count": 0, "top_ups_volume": 0.0, "last_activity": state.last_seen.isoformat() if state.last_seen else None, "pending_real_trades_count": 0, "pending_real_trades_volume": 0.0, "individual_trades": [], "individual_top_ups": [], "individual_cancellations": [], } return { "timestamp": datetime.now().isoformat(), # Используем размер итоговой мапы, включающей активных мерчантов без событий "total_merchants": len(merchant_stats_json), "active_d_variables": len(self.d_variables), "active_merchant_states": len(self.merchant_states), "processed_transactions": len(self.processed_transactions), "merchant_stats": merchant_stats_json, # Экспорт активных D‑переменных (для диагностики): "d_variables": { d_id: { "merchant_id": d.merchant_id, "merchant_name": d.merchant_name, "asset": d.asset, "fiat": d.fiat, "trade_type": d.trade_type, "amount": float(d.amount), "price_at_d1": d.price_at_d1, "timestamp": d.timestamp.isoformat(), "last_confirmation_at": d.last_confirmation_at.isoformat() if d.last_confirmation_at else None, "confirmation_stage": d.confirmation_stage, "expires_at": d.expires_at.isoformat(), "is_matched": d.is_matched, "matched_with": d.matched_with, } for d_id, d in self.d_variables.items() } } def get_merchant_stats(self) -> Dict[str, MerchantLiquidityStats]: """Получить статистику мерчантов""" return self.merchant_stats