"""AutoPOE2 - Auto-Arbitrage & Profit-Taking Orchestrator (Doc 70 Mục 5 / Chu Kỳ 7). =================================================================================== Bộ điều phối tự động quét số dư kho tiền tệ (Stash Currency Tab & Inventory), định nghĩa các ngưỡng dự trữ an toàn (reserve thresholds), tự động sinh các lệnh thanh lý currency cấp phụ thặng dư để chốt lời sang Chaos Orb, gom Divine Orb và tiến tới Mirror of Kalandra trên sàn Faustus Currency Exchange (phím '/'). 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. - Rule 5: Báo cáo gắn nhãn [MÔ PHỎNG THUẬT TOÁN HEADLESS - KHÔNG PHẢI LIVE GAME]. - Rule 7: Phím '/' mở Currency Exchange, Spacebar đóng panels. - Bất biến: INV-EXCHANGE-GOLD-TAX, INV-EXCHANGE-RESERVE-SAFE, INV-EXCHANGE-SLIPPAGE-GUARD. """ from __future__ import annotations import math import uuid from dataclasses import dataclass, field from datetime import datetime from enum import Enum from typing import Any, Dict, List, Optional, Set, Tuple from src.assistant_tool.currency_exchange.models import ( CurrencyTier, ExchangeActionType, ExchangeOrder, ExchangeOrderStatus, ExchangePortfolioConfig, ) @dataclass class ProfitTakingThresholds: """Định mức ngưỡng dự trữ tối thiểu bắt buộc giữ lại cho từng loại tiền tệ.""" reserves: Dict[str, int] = field(default_factory=lambda: { "Orb of Transmutation": 40, "Orb of Augmentation": 40, "Orb of Alteration": 40, "Orb of Alchemy": 20, "Orb of Scouring": 20, "Regal Orb": 10, "Exalted Orb": 5, "Vaul Orb": 10, "Gemcutter's Prism": 5, "Greater Jeweller's Orb": 10, "Chaos Orb": 50, "Divine Orb": 2, }) # Số lượng tối thiểu mỗi đợt bán để tránh spam lệnh vụn vặt min_liquidation_units: Dict[str, int] = field(default_factory=lambda: { "Orb of Transmutation": 20, "Orb of Augmentation": 20, "Orb of Alteration": 20, "Orb of Alchemy": 10, "Orb of Scouring": 10, "Regal Orb": 5, "Exalted Orb": 1, "Vaul Orb": 5, "Gemcutter's Prism": 2, "Greater Jeweller's Orb": 5, }) def get_reserve(self, currency_name: str) -> int: return self.reserves.get(currency_name, 0) def get_min_liquidation(self, currency_name: str) -> int: return self.min_liquidation_units.get(currency_name, 1) @dataclass class EconomicRisk: """Mô tả rủi ro kinh tế trong quá trình thanh lý & chốt lời.""" risk_type: str # "SLIPPAGE_RISK", "GOLD_DEPLETION_RISK", "CRAFTING_STARVATION_RISK" severity: str # "LOW", "MEDIUM", "HIGH", "CRITICAL" description: str mitigation_action: str @dataclass class ProfitTakingPlan: """Kế hoạch chốt lời ngoại hối tổng thể.""" total_orders: int orders: List[ExchangeOrder] projected_chaos_spent: float projected_chaos_earned: float projected_divine_bought: int projected_mirror_bought: int total_estimated_gold_tax: int remaining_gold_balance: int risks: List[EconomicRisk] status_notes: List[str] timestamp: str = field(default_factory=lambda: datetime.now().isoformat()) class AutoProfitTaker: """Bộ điều phối tự động chốt lời và phân tầng tài sản ngoại hối.""" def __init__( self, config: Optional[ExchangePortfolioConfig] = None, thresholds: Optional[ProfitTakingThresholds] = None, market_prices: Optional[Dict[str, float]] = None, ): self.config = config or ExchangePortfolioConfig() self.thresholds = thresholds or ProfitTakingThresholds() self.market_prices: Dict[str, float] = market_prices or { "Chaos Orb": 1.0, "Divine Orb": 145.0, "Mirror of Kalandra": 65250.0, "Exalted Orb": 12.5, "Regal Orb": 1.2, "Orb of Alchemy": 0.5, "Orb of Scouring": 0.8, "Vaul Orb": 1.5, "Orb of Annulment": 4.0, "Gemcutter's Prism": 3.0, "Greater Jeweller's Orb": 2.5, "Orb of Transmutation": 0.05, "Orb of Alteration": 0.15, "Orb of Augmentation": 0.05, } def scan_holdings( self, wallet: Dict[str, int], stash_currency: Optional[Dict[str, int]] = None, ) -> Dict[str, int]: """Hợp nhất toàn bộ số dư tiền tệ từ túi đồ (Inventory) và hòm đồ (Stash).""" combined: Dict[str, int] = dict(wallet) if stash_currency: for item, qty in stash_currency.items(): combined[item] = combined.get(item, 0) + qty return combined def _get_gold_tax_per_unit(self, currency_name: str) -> int: """Ước tính mức thuế vàng của Faustus cho từng loại tiền tệ.""" if currency_name in ("Orb of Transmutation", "Orb of Augmentation", "Orb of Alteration"): return 5 elif currency_name in ("Orb of Alchemy", "Orb of Scouring", "Regal Orb", "Vaul Orb"): return 25 elif currency_name in ("Exalted Orb", "Gemcutter's Prism", "Greater Jeweller's Orb"): return 50 elif currency_name == "Chaos Orb": return 20 elif currency_name == "Divine Orb": return 250 elif currency_name == "Mirror of Kalandra": return 10000 return 10 def evaluate_profit_taking( self, combined_holdings: Dict[str, int], current_gold: int, dynamic_prices: Optional[Dict[str, float]] = None, ) -> ProfitTakingPlan: """ Đánh giá số dư tổng hợp và xây dựng kế hoạch chốt lời tối ưu: 1. Kiểm tra khóa ngân sách Vàng tối thiểu (INV-EXCHANGE-GOLD-TAX). 2. Thanh lý các loại currency phụ thặng dư đổi sang Chaos Orb. 3. Gom Chaos Orb tích lũy đủ để mua Divine Orb. 4. Gom Divine Orb tích lũy đủ để mua Mirror of Kalandra. 5. Phân tích ma trận rủi ro kinh tế (Slippage, Depletion, Starvation). """ prices = dict(self.market_prices) if dynamic_prices: prices.update(dynamic_prices) notes: List[str] = [] risks: List[EconomicRisk] = [] orders: List[ExchangeOrder] = [] # 1. Khóa an toàn Vàng (INV-EXCHANGE-GOLD-TAX) available_gold = current_gold - self.config.min_gold_reserve if available_gold <= 0: notes.append( f"Khoa an toan Vang: Quy vang ({current_gold}) <= nguong bao ve " f"({self.config.min_gold_reserve}). Tu choi sinh lenh chot loi." ) risks.append(EconomicRisk( risk_type="GOLD_DEPLETION_RISK", severity="CRITICAL", description=f"Quy vang {current_gold} khong du nguong toi thieu {self.config.min_gold_reserve}.", mitigation_action="Tiep tuc farm maps tich luy vang truoc khi mo san.", )) return ProfitTakingPlan( total_orders=0, orders=[], projected_chaos_spent=0.0, projected_chaos_earned=0.0, projected_divine_bought=0, projected_mirror_bought=0, total_estimated_gold_tax=0, remaining_gold_balance=current_gold, risks=risks, status_notes=notes, ) # 2. Tầng 1: Sinh lệnh bán thặng dư currency phụ lấy Chaos Orb sell_orders: List[ExchangeOrder] = [] projected_chaos_from_sales = 0.0 for item_name, qty in combined_holdings.items(): if item_name in ("Chaos Orb", "Divine Orb", "Mirror of Kalandra", "Gold"): continue min_res = self.thresholds.get_reserve(item_name) min_liq = self.thresholds.get_min_liquidation(item_name) surplus = qty - min_res if surplus >= min_liq: rate = prices.get(item_name, 0.0) expected_chaos = int(surplus * rate) if expected_chaos >= 1: tax = surplus * self._get_gold_tax_per_unit(item_name) order = ExchangeOrder( order_id=f"order_sell_{uuid.uuid4().hex[:6]}", action_type=ExchangeActionType.SELL_TO_CHAOS, sell_currency=item_name, sell_amount=surplus, buy_currency="Chaos Orb", expected_buy_amount=expected_chaos, market_rate=rate, estimated_gold_tax=tax, status=ExchangeOrderStatus.PENDING, ) sell_orders.append(order) projected_chaos_from_sales += expected_chaos # Sắp xếp ưu tiên lệnh có giá trị Chaos kỳ vọng cao nhất lên trước sell_orders.sort(key=lambda o: o.expected_buy_amount, reverse=True) # 3. Tầng 2: Gom đổi Chaos Orb sang Divine Orb divine_price = prices.get("Divine Orb", 145.0) current_chaos = combined_holdings.get("Chaos Orb", 0) chaos_reserve = self.thresholds.get_reserve("Chaos Orb") surplus_chaos_now = max(0, current_chaos - chaos_reserve) total_available_chaos = surplus_chaos_now + projected_chaos_from_sales divine_to_buy = int(total_available_chaos // divine_price) divine_orders: List[ExchangeOrder] = [] chaos_spent = 0.0 if divine_to_buy >= 1: chaos_cost = int(divine_to_buy * divine_price) divine_tax = divine_to_buy * self._get_gold_tax_per_unit("Divine Orb") divine_order = ExchangeOrder( order_id=f"order_buy_divine_{uuid.uuid4().hex[:6]}", action_type=ExchangeActionType.CONVERT_CHAOS_TO_DIVINE, sell_currency="Chaos Orb", sell_amount=chaos_cost, buy_currency="Divine Orb", expected_buy_amount=divine_to_buy, market_rate=1.0 / divine_price, estimated_gold_tax=divine_tax, status=ExchangeOrderStatus.PENDING, ) divine_orders.append(divine_order) chaos_spent = float(chaos_cost) # 4. Tầng 3: Gom đổi Divine Orb sang Mirror of Kalandra (Apex Asset) mirror_price_in_divine = 450.0 # Chuẩn quy đổi thị trường POE2 current_divine = combined_holdings.get("Divine Orb", 0) divine_reserve = self.thresholds.get_reserve("Divine Orb") surplus_divine_now = max(0, current_divine - divine_reserve) total_potential_divine = surplus_divine_now + divine_to_buy mirror_to_buy = int(total_potential_divine // mirror_price_in_divine) mirror_orders: List[ExchangeOrder] = [] if mirror_to_buy >= 1: divine_cost = int(mirror_to_buy * mirror_price_in_divine) mirror_tax = mirror_to_buy * self._get_gold_tax_per_unit("Mirror of Kalandra") mirror_order = ExchangeOrder( order_id=f"order_buy_mirror_{uuid.uuid4().hex[:6]}", action_type=ExchangeActionType.CONVERT_DIVINE_TO_MIRROR, sell_currency="Divine Orb", sell_amount=divine_cost, buy_currency="Mirror of Kalandra", expected_buy_amount=mirror_to_buy, market_rate=1.0 / mirror_price_in_divine, estimated_gold_tax=mirror_tax, status=ExchangeOrderStatus.PENDING, ) mirror_orders.append(mirror_order) # 5. Phân bổ ngân sách Vàng và cắt tỉa lệnh nếu vượt hạn mức # Thứ tự ưu tiên lệnh: Bán thặng dư -> Mua Divine -> Mua Mirror candidate_orders = sell_orders + divine_orders + mirror_orders approved_orders: List[ExchangeOrder] = [] accumulated_gold_tax = 0 for ord_item in candidate_orders: if (accumulated_gold_tax + ord_item.estimated_gold_tax) <= available_gold: approved_orders.append(ord_item) accumulated_gold_tax += ord_item.estimated_gold_tax else: ord_item.status = ExchangeOrderStatus.REJECTED_LOW_GOLD ord_item.reason = "Vuot han muc ngan sach Vang sau khi tru min_gold_reserve" notes.append(f"Lenh {ord_item.order_id} bi cat tia do thieu han muc Vang.") # 6. Phân tích ma trận rủi ro kinh tế (Economic Risk Matrix) # Rủi ro 1: Crafting Starvation for craft_item in ("Orb of Alchemy", "Orb of Scouring"): owned = combined_holdings.get(craft_item, 0) res = self.thresholds.get_reserve(craft_item) if owned <= res: risks.append(EconomicRisk( risk_type="CRAFTING_STARVATION_RISK", severity="LOW", description=f"{craft_item} hien co {owned} sat nguong du tru ({res}). Khong the thanh ly them.", mitigation_action="Bao toan 100% so luong con lai de craft Waystone map.", )) # Rủi ro 2: Gold Depletion tax_ratio = accumulated_gold_tax / max(1, current_gold) if tax_ratio > 0.5: risks.append(EconomicRisk( risk_type="GOLD_DEPLETION_RISK", severity="MEDIUM", description=f"Thue vang chiem {tax_ratio*100:.1f}% tong quy vang hien co ({accumulated_gold_tax}/{current_gold}).", mitigation_action="Uu tien chay cac map co mod Kalguuran Taxes de hoi phuc vang.", )) remaining_gold = current_gold - accumulated_gold_tax return ProfitTakingPlan( total_orders=len(approved_orders), orders=approved_orders, projected_chaos_spent=chaos_spent, projected_chaos_earned=projected_chaos_from_sales, projected_divine_bought=sum(o.expected_buy_amount for o in approved_orders if o.action_type == ExchangeActionType.CONVERT_CHAOS_TO_DIVINE), projected_mirror_bought=sum(o.expected_buy_amount for o in approved_orders if o.action_type == ExchangeActionType.CONVERT_DIVINE_TO_MIRROR), total_estimated_gold_tax=accumulated_gold_tax, remaining_gold_balance=remaining_gold, risks=risks, status_notes=notes, ) def format_ascii_plan(self, plan: ProfitTakingPlan) -> str: """Định dạng kế hoạch chốt lời thành bảng báo cáo ASCII an toàn cho console.""" lines = [ "=" * 78, " [MO PHONG THUAT TOAN HEADLESS - KHONG PHAI LIVE GAME]", " KE HOACH CHOT LOI & TOI UU DANH MUC NGOAI HOI (AUTO PROFIT-TAKING)", "=" * 78, f" Tong So Lenh Chap Thuan : {plan.total_orders}", f" Chaos Du Kien Thu Ve : +{plan.projected_chaos_earned:,.1f} Chaos", f" Chaos Du Kien Chi Tieu : -{plan.projected_chaos_spent:,.1f} Chaos", f" Divine Du Kien Mua Duoc : +{plan.projected_divine_bought} Divine Orb", f" Mirror Du Kien Mua Duoc : +{plan.projected_mirror_bought} Mirror of Kalandra", f" Tong Thue Vang Uoc Tinh : {plan.total_estimated_gold_tax:,} Gold", f" So Du Vang Sau Giao Dich : {plan.remaining_gold_balance:,} Gold", "-" * 78, " CHI TIET CAC LENH GIAO DICH UY THAC:", ] if not plan.orders: lines.append(" (Khong co lenh nao du dieu kien hoac quy vang khong du)") else: for idx, o in enumerate(plan.orders, start=1): lines.append( f" {idx:>2}. [{o.action_type.value:<24}] Ban {o.sell_amount:>5} {o.sell_currency:<22} " f"-> Mua {o.expected_buy_amount:>4} {o.buy_currency:<18} (Thue: {o.estimated_gold_tax:>5} Gold)" ) if plan.risks: lines.extend([ "-" * 78, " PHAN TICH MA TRAN RUI RO KINH TE (ECONOMIC RISK MATRIX):", ]) for r in plan.risks: lines.append(f" * [{r.severity}] {r.risk_type}: {r.description}") lines.append(f" -> Bien phap: {r.mitigation_action}") lines.append("=" * 78) return "\n".join(lines)