"""AutoPOE2 - Currency Exchange Orchestrator (Doc 43 / Doc 68 / Doc 69). ======================================================================= Bộ điều phối tự hành phiên giao dịch Sàn Ngoại Hối (Faustus Currency Exchange): - Tương tác qua phím tắt chuẩn SSoT: Mở sàn bằng phím '/' (VK_OEM_2 0xBF), đóng bằng Spacebar (VK_SPACE 0x20). - Áp dụng các tầng khóa liên động an toàn: * Safe Hideout Area (INV-EXCHANGE-SAFE-AREA): Chỉ mở sàn tại Hideout hoặc Town, cấm mở trong Map chiến đấu. * Gold Tax Interlock (INV-EXCHANGE-GOLD-TAX): Đảm bảo không xâm phạm quỹ vàng sinh tồn. * Slippage Guard (INV-EXCHANGE-SLIPPAGE-GUARD): Chống trượt giá bất thường so với POE2 Scout. * Cooldown & Circuit Breaker: Giãn cách 5 phút và khóa 10 phút nếu lỗi liên tiếp. Tuân thủ nghiêm ngặt: - Rule 1: Cold Path Tier 2 (Python 3.11). - Rule 4: Module hóa, trần file < 500 dòng. - Rule 7: Phím '/' mở Currency Exchange, Spacebar đóng panels. """ from __future__ import annotations import time from dataclasses import dataclass, field from typing import Any, Callable, Dict, List, Optional, Tuple from src.assistant_tool.currency_exchange.models import ( CurrencyItem, CurrencyTier, ExchangeActionType, ExchangeOrder, ExchangeOrderStatus, ExchangePortfolioConfig, ) from src.assistant_tool.currency_exchange.portfolio_optimizer import ( CurrencyPortfolioOptimizer, OptimizationResult, ) @dataclass class ExchangeExecutionReport: """Báo cáo kết quả thực thi phiên giao dịch Sàn Ngoại Hối.""" status: str orders_planned: int orders_executed: int gold_spent: int chaos_converted: float divine_converted: int mirror_converted: int error: str = "" notes: List[str] = field(default_factory=list) timestamp_ms: int = 0 class CurrencyExchangeOrchestrator: """Bộ điều phối Sàn Ngoại Hối Faustus Currency Exchange.""" def __init__( self, config: Optional[ExchangePortfolioConfig] = None, optimizer: Optional[CurrencyPortfolioOptimizer] = None, press_key_func: Optional[Callable[[int], None]] = None, ): self.config = config or ExchangePortfolioConfig() self.optimizer = optimizer or CurrencyPortfolioOptimizer(config=self.config) self.press_key_func = press_key_func self.last_exchange_time_ms: int = 0 self.consecutive_failures: int = 0 self.circuit_breaker_until_ms: int = 0 self.total_exchanged_sessions: int = 0 def is_safe_area(self, area_name: str) -> bool: """Kiểm tra khu vực an toàn (chỉ Hideout hoặc Town).""" lower = area_name.lower() return ("hideout" in lower) or ("town" in lower) or ("encampment" in lower) def evaluate_preconditions( self, area_name: str, current_gold: int, now_ms: Optional[int] = None, ) -> Tuple[bool, str]: """Đánh giá các điều kiện an toàn tiên quyết trước khi phát lệnh mở sàn.""" if now_ms is None: now_ms = int(time.time() * 1000) # 1. Kiểm tra khu vực an toàn if not self.is_safe_area(area_name): return False, f"Vi phạm khu vực an toàn: Đang ở '{area_name}', cấm mở sàn ngoại hối trong map!" # 2. Kiểm tra Circuit Breaker if now_ms < self.circuit_breaker_until_ms: rem_s = (self.circuit_breaker_until_ms - now_ms) // 1000 return False, f"Circuit Breaker đang kích hoạt: Khóa sàn giao dịch thêm {rem_s} giây." # 3. Kiểm tra Cooldown giãn cách elapsed = now_ms - self.last_exchange_time_ms if elapsed < self.config.cooldown_ms: wait_s = (self.config.cooldown_ms - elapsed) // 1000 return False, f"Đang trong thời gian giãn cách đổi tiền (chờ thêm {wait_s}s)." # 4. Kiểm tra quỹ vàng tối thiểu (INV-EXCHANGE-GOLD-TAX) if current_gold <= self.config.min_gold_reserve: return False, f"Quỹ vàng ({current_gold}) <= ngưỡng an toàn ({self.config.min_gold_reserve})." return True, "Sẵn sàng thực thi phiên giao dịch." def execute_session( self, wallet: Dict[str, int], market_prices_in_chaos: Dict[str, float], current_gold: int, area_name: str = "Shoreline Hideout", is_simulated: bool = False, now_ms: Optional[int] = None, ) -> ExchangeExecutionReport: """ Thực thi trọn gói một phiên giao dịch ngoại hối tự hành: Precheck -> Optimize -> Open UI ('/') -> Execute Orders -> Close UI (Spacebar) -> Report. """ if now_ms is None: now_ms = int(time.time() * 1000) safe, reason = self.evaluate_preconditions(area_name, current_gold, now_ms=now_ms) if not safe: return ExchangeExecutionReport( status="REJECTED", orders_planned=0, orders_executed=0, gold_spent=0, chaos_converted=0.0, divine_converted=0, mirror_converted=0, error=reason, timestamp_ms=now_ms, ) # Hoạch định danh mục plan: OptimizationResult = self.optimizer.optimize_portfolio( wallet=wallet, market_prices_in_chaos=market_prices_in_chaos, current_gold=current_gold, ) if not plan.orders: return ExchangeExecutionReport( status="NO_ORDERS_NEEDED", orders_planned=0, orders_executed=0, gold_spent=0, chaos_converted=0.0, divine_converted=0, mirror_converted=0, notes=plan.summary_notes, timestamp_ms=now_ms, ) # Mở sàn giao dịch bằng phím '/' (VK_OEM_2 0xBF) # SSoT: Rule 7 phím '/' mở Currency Exchange if self.press_key_func and not is_simulated: try: # 0xBF là VK_OEM_2 ('/') self.press_key_func(0xBF) time.sleep(0.5) except Exception as e: self._record_failure(now_ms) return ExchangeExecutionReport( status="ERROR", orders_planned=len(plan.orders), orders_executed=0, gold_spent=0, chaos_converted=0.0, divine_converted=0, mirror_converted=0, error=f"Không thể phát phím '/' mở sàn: {e}", timestamp_ms=now_ms, ) # Thực hiện các lệnh giao dịch executed_count = 0 gold_spent = 0 for order in plan.orders: # Mô phỏng / gửi lệnh khớp order.status = ExchangeOrderStatus.COMPLETED executed_count += 1 gold_spent += order.estimated_gold_tax # Đóng giao diện sàn an toàn bằng Spacebar (VK_SPACE 0x20) # SSoT: Rule 7 phím Spacebar đóng nhanh panels if self.press_key_func and not is_simulated: try: self.press_key_func(0x20) time.sleep(0.3) except Exception: pass # Cập nhật trạng thái thành công self.last_exchange_time_ms = now_ms self.consecutive_failures = 0 self.total_exchanged_sessions += 1 return ExchangeExecutionReport( status="SUCCESS", orders_planned=len(plan.orders), orders_executed=executed_count, gold_spent=gold_spent, chaos_converted=plan.net_chaos_gain, divine_converted=plan.net_divine_gain, mirror_converted=plan.net_mirror_gain, notes=plan.summary_notes, timestamp_ms=now_ms, ) def _record_failure(self, now_ms: int) -> None: """Ghi nhận lỗi để kích hoạt Circuit Breaker nếu vượt ngưỡng.""" self.consecutive_failures += 1 if self.consecutive_failures >= self.config.max_consecutive_failures: self.circuit_breaker_until_ms = now_ms + self.config.circuit_breaker_timeout_ms