"""AutoPOE2 - Faustus Currency Arbitrage & Optimal Conversion Engine (Doc 68 / Doc 70). ======================================================================================== Mô hình hóa thị trường tiền tệ ngoại hối Faustus (phím '/') dưới dạng đồ thị có hướng (Directed Weighted Graph). - Thuật toán Dijkstra / Bellman-Ford tìm đường chuyển đổi tiền tệ có tỷ giá tối ưu nhất (Best Route). - Phát hiện cơ hội Arbitrage tam giác (Triangular Arbitrage) tích lũy tài sản thụ động giữa các map. - Cơ chế trượt giá an toàn (Slippage Guard <= 3.0%) và TTL lệnh 180s (1 Map). - Tối ưu hóa chi phí thu mua Waystone T15-T16 và thanh lý ngoại tệ thặng dư sang Divine/Chaos. Tuân thủ nghiêm ngặt: - Rule 1: Cold Path Tier 2 (Python 3.11). - Rule 4: Module hóa độc lập, trần file < 500 dòng (Anti-Monolithic). - Rule 5: Empirical DoD & Authentic Logic. - Bất biến: INV-EXCHANGE-GOLD-TAX, INV-EXCHANGE-RESERVE-SAFE, INV-EXCHANGE-SLIPPAGE-GUARD. """ from __future__ import annotations import heapq import math from dataclasses import dataclass, field from datetime import datetime from typing import Any, Dict, List, Optional, Set, Tuple from src.assistant_tool.currency_exchange.models import ( CurrencyTier, ExchangeActionType, ExchangeOrder, ExchangeOrderStatus, ExchangePortfolioConfig, ) @dataclass class CurrencyEdge: """Cạnh trên đồ thị tỷ giá: đại diện cho 1 chiều giao dịch từ from_curr sang to_curr.""" from_curr: str to_curr: str rate: float # Số đơn vị to_curr nhận được trên 1 đơn vị from_curr gold_tax_per_unit: int = 50 # Thuế vàng trên mỗi đơn vị from_curr bid_spread: float = 0.01 # Spread mua/bán thị trường (1%) @property def effective_rate(self) -> float: """Tỷ giá thực tế sau khi trừ spread thị trường.""" return self.rate * (1.0 - self.bid_spread) @property def log_weight(self) -> float: """Trọng số logarit âm để chuyển bài toán nhân tỷ giá thành bài toán đường đi ngắn nhất.""" if self.effective_rate <= 0: return float("inf") return -math.log(self.effective_rate) @dataclass class ConversionHop: """Một chặng giao dịch trong tuyến đường chuyển đổi.""" from_curr: str to_curr: str rate: float amount_in: float amount_out: float gold_tax: int @dataclass class ConversionRoute: """Tuyến đường chuyển đổi tỷ giá tối ưu.""" source_curr: str target_curr: str hops: List[ConversionHop] final_amount: float effective_rate: float direct_rate: Optional[float] = None rate_advantage_pct: float = 0.0 # Chênh lệch có lợi so với giao dịch trực tiếp (%) total_gold_tax: int = 0 estimated_slippage_pct: float = 0.0 is_arbitrage_profitable: bool = False @property def path_str(self) -> str: if not self.hops: return self.source_curr return " -> ".join([self.hops[0].from_curr] + [h.to_curr for h in self.hops]) @dataclass class ArbitrageOpportunity: """Cơ hội Arbitrage tam giác có lãi sau thuế Vàng.""" cycle_path: List[str] gross_profit_pct: float net_profit_chaos: float total_gold_tax: int is_executable: bool reason: str = "" @dataclass class LiquidationPlan: """Kế hoạch thanh lý ngoại tệ thặng dư sang Divine / Chaos.""" total_orders: int orders: List[ExchangeOrder] projected_chaos_earned: float projected_divine_earned: int total_gold_tax: int remaining_gold_balance: int source_currencies: List[str] class FaustusCurrencyArbitrage: """ Hạt nhân phân tích tỷ giá & điều phối chênh lệch giá sàn Faustus. """ def __init__( self, config: Optional[ExchangePortfolioConfig] = None, max_slippage_pct: float = 3.0, default_ttl_sec: float = 180.0, min_gold_reserve: int = 5000, chaos_per_gold_rate: float = 0.0002, # 1 Gold ~ 0.0002 Chaos (5,000 Gold = 1 Chaos) ): self.config = config or ExchangePortfolioConfig() self.max_slippage_pct = max_slippage_pct self.default_ttl_sec = default_ttl_sec self.min_gold_reserve = min_gold_reserve self.chaos_per_gold_rate = chaos_per_gold_rate # Đồ thị tỷ giá: graph[from_curr] = List[CurrencyEdge] self.graph: Dict[str, List[CurrencyEdge]] = {} def add_rate( self, from_curr: str, to_curr: str, rate: float, gold_tax: int = 50, bid_spread: float = 0.01, ) -> None: """Thêm hoặc cập nhật tỷ giá trực tiếp giữa 2 loại tiền tệ.""" if from_curr not in self.graph: self.graph[from_curr] = [] # Xóa cạnh cũ nếu tồn tại self.graph[from_curr] = [e for e in self.graph[from_curr] if e.to_curr != to_curr] edge = CurrencyEdge( from_curr=from_curr, to_curr=to_curr, rate=rate, gold_tax_per_unit=gold_tax, bid_spread=bid_spread, ) self.graph[from_curr].append(edge) def load_rates_from_market_pairs( self, market_pairs: Dict[Tuple[str, str], float], gold_tax_map: Optional[Dict[str, int]] = None, ) -> None: """Nạp danh mục tỷ giá thị trường từ POE2 Scout API hoặc RAM cache.""" tax_map = gold_tax_map or {} for (from_c, to_c), rate in market_pairs.items(): if rate > 0: tax = tax_map.get(from_c, 50) self.add_rate(from_c, to_c, rate, gold_tax=tax) def find_best_conversion_path( self, source_curr: str, target_curr: str, amount_in: float = 1.0, ) -> Optional[ConversionRoute]: """ Thuật toán Dijkstra trên trọng số -log(rate) tìm đường chuyển đổi có tỷ giá đầu ra lớn nhất. Tuân thủ Slippage Guard <= 3.0%. """ if source_curr == target_curr: return ConversionRoute( source_curr=source_curr, target_curr=target_curr, hops=[], final_amount=amount_in, effective_rate=1.0, rate_advantage_pct=0.0, total_gold_tax=0, ) if source_curr not in self.graph: return None # Min-Heap cho Dijkstra: (cumulative_log_weight, current_currency, path_edges) heap = [(0.0, source_curr, [])] best_weights: Dict[str, float] = {source_curr: 0.0} best_edges_path: Optional[List[CurrencyEdge]] = None while heap: cur_weight, cur_curr, cur_path = heapq.heappop(heap) if cur_weight > best_weights.get(cur_curr, float("inf")): continue if cur_curr == target_curr: best_edges_path = cur_path break for edge in self.graph.get(cur_curr, []): new_weight = cur_weight + edge.log_weight if new_weight < best_weights.get(edge.to_curr, float("inf")): best_weights[edge.to_curr] = new_weight heapq.heappush(heap, (new_weight, edge.to_curr, cur_path + [edge])) if not best_edges_path: return None # Xây dựng ConversionRoute chi tiết hops: List[ConversionHop] = [] cur_amount = amount_in total_gold_tax = 0 for edge in best_edges_path: hop_amount_out = cur_amount * edge.effective_rate hop_gold_tax = int(cur_amount * edge.gold_tax_per_unit) hops.append(ConversionHop( from_curr=edge.from_curr, to_curr=edge.to_curr, rate=edge.rate, amount_in=cur_amount, amount_out=hop_amount_out, gold_tax=hop_gold_tax, )) total_gold_tax += hop_gold_tax cur_amount = hop_amount_out effective_rate = cur_amount / amount_in if amount_in > 0 else 0.0 # Tìm tỷ giá trực tiếp để đối soát chênh lệch lợi nhuận direct_rate = None for edge in self.graph.get(source_curr, []): if edge.to_curr == target_curr: direct_rate = edge.effective_rate break advantage_pct = 0.0 if direct_rate and direct_rate > 0: advantage_pct = round(((effective_rate / direct_rate) - 1.0) * 100.0, 2) # Kiểm tra trượt giá dự tính (ước tính dựa trên độ dài hops x spread 1%) slippage_est = len(hops) * 1.0 if slippage_est > self.max_slippage_pct: return None # Vi phạm Slippage Guard <= 3.0% return ConversionRoute( source_curr=source_curr, target_curr=target_curr, hops=hops, final_amount=cur_amount, effective_rate=effective_rate, direct_rate=direct_rate, rate_advantage_pct=advantage_pct, total_gold_tax=total_gold_tax, estimated_slippage_pct=slippage_est, is_arbitrage_profitable=(advantage_pct > 0.5), ) def detect_triangular_arbitrage( self, base_currency: str = "Chaos Orb", test_amount: float = 100.0, current_gold: int = 15000, ) -> List[ArbitrageOpportunity]: """ Tìm kiếm các chu trình Arbitrage tam giác xuất phát và quay về base_currency. Tự động khấu trừ chi phí thuế Vàng để xác định lợi nhuận ròng thực tế. """ opportunities: List[ArbitrageOpportunity] = [] if base_currency not in self.graph: return opportunities # Duyệt qua các chu trình độ dài 3 (Base -> A -> B -> Base) for e1 in self.graph[base_currency]: curr_a = e1.to_curr if curr_a == base_currency: continue for e2 in self.graph.get(curr_a, []): curr_b = e2.to_curr if curr_b in (base_currency, curr_a): continue for e3 in self.graph.get(curr_b, []): if e3.to_curr == base_currency: # Chu trình hoàn chỉnh: base -> a -> b -> base step1 = test_amount * e1.effective_rate step2 = step1 * e2.effective_rate final_chaos = step2 * e3.effective_rate gross_profit = final_chaos - test_amount gross_profit_pct = (gross_profit / test_amount) * 100.0 total_tax_gold = int( test_amount * e1.gold_tax_per_unit + step1 * e2.gold_tax_per_unit + step2 * e3.gold_tax_per_unit ) tax_cost_in_chaos = total_tax_gold * self.chaos_per_gold_rate net_profit_chaos = gross_profit - tax_cost_in_chaos # Điều kiện an toàn vốn & quỹ vàng can_execute = True reason = "Arbitrage co lai rong sau thue Vang." if current_gold - total_tax_gold < self.min_gold_reserve: can_execute = False reason = f"Quy vang sau giao dich ({current_gold - total_tax_gold}) thap hon nguong toi thieu ({self.min_gold_reserve})." elif net_profit_chaos <= 1.0: can_execute = False reason = f"Loi nhuan rong ({net_profit_chaos:.2f} Chaos) khong bu du chi phi truot gia va thue Vang." opp = ArbitrageOpportunity( cycle_path=[base_currency, curr_a, curr_b, base_currency], gross_profit_pct=round(gross_profit_pct, 2), net_profit_chaos=round(net_profit_chaos, 2), total_gold_tax=total_tax_gold, is_executable=can_execute, reason=reason, ) opportunities.append(opp) # Sắp xếp cơ hội theo lợi nhuận ròng giảm dần opportunities.sort(key=lambda x: x.net_profit_chaos, reverse=True) return opportunities def plan_waystone_purchase( self, target_tier: int = 16, count: int = 5, available_wallet: Optional[Dict[str, int]] = None, current_gold: int = 15000, ) -> Optional[ConversionRoute]: """ Tìm tuyến đường mua Waystone với chi phí quy đổi thấp nhất từ kho tài sản hiện có. """ wallet = available_wallet or {} waystone_key = f"Waystone (Tier {target_tier})" best_route: Optional[ConversionRoute] = None lowest_cost_in_chaos = float("inf") # Khảo sát mua qua Chaos Orb hoặc Divine Orb for payment_curr in ["Chaos Orb", "Divine Orb", "Exalted Orb"]: avail = wallet.get(payment_curr, 0) reserve = self.config.default_reserves.get(payment_curr, 0) spendable = max(0, avail - reserve) if spendable <= 0: continue route = self.find_best_conversion_path(payment_curr, waystone_key, amount_in=float(spendable)) if route and route.final_amount >= count: # Tính chi phí quy đổi sang Chaos chuẩn cost_in_chaos = (count / (route.final_amount / spendable)) if payment_curr == "Divine Orb": cost_in_chaos *= 145.0 # Tỷ giá Divine/Chaos tham chiếu if cost_in_chaos < lowest_cost_in_chaos: lowest_cost_in_chaos = cost_in_chaos best_route = route return best_route def plan_surplus_liquidation( self, wallet: Dict[str, int], current_gold: int = 15000, target_currency: str = "Divine Orb", ) -> LiquidationPlan: """ Lập kế hoạch thanh lý toàn bộ ngoại tệ thặng dư vượt định mức dự trữ sang target_currency. Tuân thủ INV-EXCHANGE-RESERVE-SAFE và INV-EXCHANGE-GOLD-TAX. """ orders: List[ExchangeOrder] = [] total_chaos_earned = 0.0 total_gold_tax = 0 rem_gold = current_gold source_currencies: List[str] = [] reserves = self.config.default_reserves for curr_name, quantity in wallet.items(): if curr_name in (target_currency, "Mirror of Kalandra", "Chaos Orb"): continue min_res = reserves.get(curr_name, 10) surplus = quantity - min_res if surplus <= 0: continue route = self.find_best_conversion_path(curr_name, "Chaos Orb", amount_in=float(surplus)) if not route: continue if rem_gold - route.total_gold_tax < self.min_gold_reserve: break # Chạm ngưỡng bảo vệ quỹ Vàng order = ExchangeOrder( order_id=f"liq_{curr_name.replace(' ', '_').lower()}", action_type=ExchangeActionType.SELL_TO_CHAOS, sell_currency=curr_name, sell_amount=surplus, buy_currency="Chaos Orb", expected_buy_amount=int(route.final_amount), market_rate=route.effective_rate, estimated_gold_tax=route.total_gold_tax, status=ExchangeOrderStatus.PENDING, ) orders.append(order) total_chaos_earned += route.final_amount total_gold_tax += route.total_gold_tax rem_gold -= route.total_gold_tax source_currencies.append(curr_name) # Gom Chaos Orb mua Divine Orb nếu tích lũy đủ divine_earned = 0 divine_rate = 1.0 / 145.0 # 1 Divine ~ 145 Chaos for edge in self.graph.get("Chaos Orb", []): if edge.to_curr == target_currency: divine_rate = edge.effective_rate break divine_cost_chaos = 1.0 / divine_rate if divine_rate > 0 else 145.0 existing_chaos = max(0, wallet.get("Chaos Orb", 0) - reserves.get("Chaos Orb", 50)) net_chaos_pool = total_chaos_earned + existing_chaos divine_to_buy = int(net_chaos_pool // divine_cost_chaos) if divine_to_buy >= 1: divine_tax = divine_to_buy * 250 if rem_gold - divine_tax >= self.min_gold_reserve: order_div = ExchangeOrder( order_id="liq_convert_chaos_to_divine", action_type=ExchangeActionType.CONVERT_CHAOS_TO_DIVINE, sell_currency="Chaos Orb", sell_amount=int(divine_to_buy * divine_cost_chaos), buy_currency=target_currency, expected_buy_amount=divine_to_buy, market_rate=divine_rate, estimated_gold_tax=divine_tax, status=ExchangeOrderStatus.PENDING, ) orders.append(order_div) divine_earned = divine_to_buy total_gold_tax += divine_tax rem_gold -= divine_tax return LiquidationPlan( total_orders=len(orders), orders=orders, projected_chaos_earned=round(total_chaos_earned, 1), projected_divine_earned=divine_earned, total_gold_tax=total_gold_tax, remaining_gold_balance=rem_gold, source_currencies=source_currencies, )