Python量化机器人实战:从自适应策略到富途模拟盘交易

class AdaptiveStrategy:    def __init__(self, threshold_vol=15.0, threshold_adx=20.0):        self.strategies = {            'mean_reversion': MeanReversionStrategy(),            'trend_following': TrendFollowingStrategy(),            'momentum': MomentumStrategy(),            'event_driven': EventDrivenStrategy()        }    def select_strategy(self, all_codes, current_prices):        # 计算市场波动率        prices = [current_prices[c] for c in all_codes if c in current_prices]        vol = np.std(prices) / np.mean(prices) * 100        adx = 25.0 if vol > 2.0 else 15.0   # ADX简化计算        if vol < self.threshold_vol:            return 'mean_reversion' if adx < self.threshold_adx else 'trend_following'        else:            return 'event_driven' if adx < self.threshold_adx else 'momentum'    def get_signal(self, code, price, is_held, all_codes, current_prices):        selected = self.select_strategy(all_codes, current_prices)        return self.strategies[selected].get_signal(code, price, is_held)

class SimulatedAccount:    def place_order(self, code, side, shares):        order = Order(id=self._next_id(), code=code, side=side, shares=shares)        self.orders.append(order)        return orderclass MatchingEngine:    def simulate_fill(self, order, current_price):        # 市价单：买入上滑，卖出下滑        if order.side == BUY:            fill_price = current_price * (1 + self.slippage)        else:            fill_price = current_price * (1 - self.slippage)        return True, fill_price

class RiskManager:    def __init__(self, max_position_pct=0.25, max_daily_loss_pct=0.05):        self.max_position_pct = max_position_pct        self.max_daily_loss_pct = max_daily_loss_pct        self.daily_start_asset = None    def check_order(self, account, code, shares, price):        total_asset = account.get_total_asset()        if self.daily_start_asset is None:            self.daily_start_asset = total_asset        daily_loss = self.daily_start_asset - total_asset        if daily_loss > self.daily_start_asset * self.max_daily_loss_pct:            return False, "每日亏损限额触发"        order_value = shares * price        if order_value > total_asset * self.max_position_pct:            return False, f"单笔仓位超限"        return True, "ok"

pip install futu-api

from futu import OpenQuoteContext, SubType, RET_OKclass FutuRealtimeDataFeed:    def __init__(self, host='127.0.0.1', port=11111):        self.quote_ctx = OpenQuoteContext(host, port)        self.last_prices = {}    def subscribe(self, code_list):        ret, _ = self.quote_ctx.subscribe(code_list, [SubType.QUOTE])        return ret == RET_OK    def get_all_prices(self, codes):        ret, data = self.quote_ctx.get_market_snapshot(codes)        if ret == RET_OK:            return {row['code']: row['last_price'] for _, row in data.iterrows()}        return {}

from futu import OpenSecTradeContext, TrdEnv, TrdSide, OrderTypeclass FutuAccount:    def __init__(self, host='127.0.0.1', port=11111):        self.trade_ctx = OpenSecTradeContext(host, port)        self.trd_env = TrdEnv.SIMULATE   # 模拟盘，改成 REAL 即实盘    def place_order(self, code, side, shares):        trd_side = TrdSide.BUY if side == 'BUY' else TrdSide.SELL        ret, data = self.trade_ctx.place_order(            code=code, price=0.0, qty=shares,            trd_side=trd_side, order_type=OrderType.MARKET,            trd_env=self.trd_env        )        return ret == RET_OK, data

class FutuAdaptiveBot:    def run_once(self):        prices = self.data_feed.get_all_prices(self.etf_pool)        for code, price in prices.items():            is_held = self.account.is_held(code)            signal = self.strategy.get_signal(code, price, is_held,                                              self.etf_pool, prices)            if signal == 'BUY' and not is_held:                ok, msg = self.risk_mgr.check_order(...)                if ok:                    self.account.place_order(code, 'BUY', 100)