現在交易的結果有點不如預期,打算再重新寫一個交易模擬,驗證一下之前策略的正確性:
from __future__ import annotations
import abc
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Dict, List, Optional, Tuple
from datetime import datetime
from decimal import Decimal
# ========= 交易所钱包 =========
@dataclass
class ExchangeWallet:
"""交易所钱包数据结构,使用 Decimal 避免浮点数精度问题"""
usd: Decimal = field(default_factory=lambda: Decimal('0'))
def deposit(self, amount: float) -> None:
"""存入资金"""
if amount <= 0:
raise ValueError("存入金额必须大于0")
self.usd += Decimal(str(amount))
def withdraw(self, amount: float) -> bool:
"""提取资金,成功返回True,余额不足返回False"""
amount_decimal = Decimal(str(amount))
if amount_decimal <= 0:
raise ValueError("提取金额必须大于0")
if self.usd >= amount_decimal:
self.usd -= amount_decimal
return True
print(f"錢包餘額({self.usd})不足{amount_decimal},withdraw失敗 (如果已經確認過內部餘額足夠, 則表示出現了預期之外的bug)")
return False
def is_available(self, amount: float) -> bool:
"""检查是否有足够余额"""
return self.usd >= Decimal(str(amount))
def get_balance(self) -> Decimal:
"""获取当前余额"""
return self.usd
# ========= 價格數據結構 =========
@dataclass
class Price:
timestamp: datetime
open: Decimal
high: Decimal
low: Decimal
close: Decimal
# ========= 價格管理抽象介面 =========
class FutureExchangePriceManager(abc.ABC):
"""永續合約價格管理抽象介面"""
def __init__(self, interval_minutes: int = 1):
"""
初始化價格管理器
:param interval_minutes: 時間級別,以分鐘為單位 (1, 5, 15, 60等)
"""
self.interval_minutes = interval_minutes
@abc.abstractmethod
def get_current_price(self) -> Optional[Price]:
"""獲取當前價格數據"""
...
@abc.abstractmethod
def increase_time_step(self) -> bool:
"""推進時間步長,返回是否成功"""
...
@abc.abstractmethod
def get_current_close_price(self) -> Optional[Decimal]:
"""獲取當前收盤價,用於快速價格查詢"""
...
# ========= 共用資料結構 =========
class OrderSide(str, Enum):
LONG = "LONG"
CLOSE_LONG = "CLOSE_LONG" # reduce-only 平多
# 之後可加:SHORT、CLOSE_SHORT
class OrderStatus(str, Enum):
NEW = "NEW"
PARTIALLY_FILLED = "PARTIALLY_FILLED"
FILLED = "FILLED"
CANCELED = "CANCELED"
EXPIRED = "EXPIRED"
@dataclass
class StopLoss:
"""止損單數據結構"""
order_id: str # 止損單ID
symbol: str # 交易對
side: OrderSide # 訂單方向(目前只支援 CLOSE_LONG)
quantity: float # 止損數量
trigger_price: float # 觸發價格
limit_price: Optional[float] = None # 限價價格,None 表示市價
create_ts: Optional[float] = None # 創建時間
update_ts: Optional[float] = field(default=None) # 觸發時間(與 FutureLimitOrder 規則一致)
def __post_init__(self):
"""初始化後處理:如果沒有提供創建時間,使用當前時間"""
if self.create_ts is None:
self.create_ts = time.time()
@dataclass
class FundingFeeRecord:
"""資金費率收取記錄"""
symbol: str # 交易對
fee: float # 收取的費用金額(USDT)
from_position: bool # 是否從持倉保證金中收取(True)還是從可用餘額收取(False)
ts: float # 收取時間戳
position_quantity: Optional[float] = None # 收取時的持倉數量
position_price: Optional[float] = None # 收取時的價格
funding_rate: Optional[float] = None # 使用的資金費率
@dataclass
class FutureLimitOrder:
"""統一的訂單數據結構(支持開多和平多)"""
order_id: str
symbol: str
side: OrderSide # 訂單方向(LONG 或 CLOSE_LONG)
target_amount_usdt: float # 目標花費金額(USDT)
actual_amount_usdt: float # 實際交易金額(USDT,不含手續費)
limit_price: float # 限價價格
quantity: float # 下單數量
commission_asset: str # 手續費資產
commission: float # 手續費數量
status: OrderStatus = field(default=OrderStatus.NEW)
executed_qty: float = field(default=0.0)
create_ts: Optional[float] = None # 訂單創建時間
update_ts: Optional[float] = field(default=None) # 訂單最後更新時間(執行、取消等)
# 槓桿交易相關欄位
realized_pnl: Optional[float] = field(default=None) # 實現損益(僅 CLOSE_LONG 訂單)
leverage: Optional[float] = field(default=None) # 訂單使用的槓桿率
def __post_init__(self):
"""初始化後處理:如果沒有提供創建時間,使用當前時間"""
if self.create_ts is None:
self.create_ts = time.time()
@property
def ts(self) -> float:
"""向後兼容性屬性,返回創建時間"""
return self.create_ts
@property
def entry_price(self) -> Optional[float]:
"""入場價格(僅適用於 LONG 訂單)"""
if self.side != OrderSide.LONG:
raise ValueError("entry_price 僅適用於 LONG 訂單,CLOSE_LONG 訂單請使用 avg_close_price")
if self.executed_qty > 0:
return self.actual_amount_usdt / self.executed_qty
return None
@property
def avg_close_price(self) -> Optional[float]:
"""平均平倉價格(僅適用於 CLOSE_LONG 訂單)"""
if self.side != OrderSide.CLOSE_LONG:
raise ValueError("avg_close_price 僅適用於 CLOSE_LONG 訂單,LONG 訂單請使用 entry_price")
if self.executed_qty > 0:
return self.actual_amount_usdt / self.executed_qty
return None
@property
def total_cost(self) -> float:
"""實際總成本 - 僅適用於買入訂單(LONG)
槓桿交易模式:僅手續費(保證金由 position.margin 管理)
"""
if self.side == OrderSide.CLOSE_LONG:
raise ValueError("CLOSE_LONG 訂單不應使用 total_cost,請使用 net_proceeds 獲取淨收入")
# 在槓桿交易系統中,實際成本只有手續費(無論槓桿倍數)
return self.commission
@property
def net_proceeds(self) -> float:
"""實際淨收益 - 適用於賣出訂單(CLOSE_LONG)
槓桿交易模式:實現損益 - 手續費
"""
if self.side != OrderSide.CLOSE_LONG:
raise ValueError("net_proceeds 僅適用於 CLOSE_LONG 訂單,其他訂單請使用 total_cost")
# 在槓桿交易系統中,使用實現損益(如果可用)
if self.realized_pnl is not None:
return self.realized_pnl - self.commission
else:
raise ValueError("realized_pnl is None, net_proceeds()執行失敗")
@dataclass
class Position:
symbol: str
quantity: float
entry_price: float
ts: Optional[float] = None # timestamp,建立時手動提供
# --- 槓桿 / 保證金 相關(為維持相容性全部提供預設值) ---
margin: float = 0.0 # 當前倉位已佔用的初始保證金 (isolated 簡化)
margin_ratio: float = 0.0 # 維持保證金率 / 保證金餘額之比(簡化)
liquidation_price: Optional[float] = None # 強平價格(簡化)
leverage: float = 1.0 # 倉位使用的槓桿(跟隨交易所初始化)
def __post_init__(self):
"""初始化後處理:如果沒有提供時間戳,使用當前時間"""
if self.ts is None:
self.ts = time.time()
def get_unrealized_pnl(self, current_price: float) -> float:
"""計算未實現損益(不包含開倉與關倉手續費)
:param current_price: 當前價格
:return: 未實現損益
"""
return (current_price - self.entry_price) * self.quantity