用Python打造自己的量化交易系统：从均线交叉到完整策略的全流程实战

作者 | 真龙现身 · 2026年5月11日

很多散户投资者在谈到量化交易时，第一反应是"太高深了，我不懂编程"、"那是专业机构才能玩的东西"。但实际上，量化交易的核心逻辑并没有那么复杂，它的本质就是：用数据说话的选股策略，用规则替代情绪的交易系统。只要你有一定的Python基础，就完全可以自己动手搭建一套简单的量化交易系统。今天这篇文章，我将手把手教你从零开始，用Python实现一个完整的均线交叉策略，涵盖数据获取、策略回测、信号生成和实盘模拟的全流程。读完这篇文章，你将拥有一个可以立即投入实盘模拟的量化交易框架。全文7500字以上，干货满满，建议先收藏再慢慢研究。

在正式开始之前，我想先回答一个很多人都会问的问题：散户做量化真的有优势吗？我的答案是，在某些方面确实有优势。量化系统可以同时监控成百上千只股票，这是人工操作完全做不到的；量化系统可以严格止盈止损，不会因为贪婪或恐惧而错过最佳买卖点；量化系统可以回测历史数据，用数据验证策略的有效性，而不是凭感觉做决策。当然，量化交易也有它的局限性，但作为散户投资者，如果能够掌握基础的量化技能，至少可以在信息获取和交易执行这两个环节大大提升效率。接下来，让我们开始实战。

01 环境准备：Python量化开发环境搭建

Python量化交易开发环境是整个系统的基础

搭建量化交易环境是入门的第一个门槛，很多新手就是卡在这一步久攻不下，最后选择了放弃。我建议使用Anaconda来管理Python环境，因为它预装了大部分科学计算所需的库，而且环境隔离功能可以让你在不同项目之间切换时不会相互干扰。具体步骤是：第一步，下载并安装Anaconda；第二步，创建一个专门用于量化交易的虚拟环境；第三步，在虚拟环境中安装需要的第三方库。对于量化交易来说，最核心的库包括：pandas用于数据处理，numpy用于数值计算，tushare或akshare用于获取股票数据，matplotlib用于可视化回测结果，backtrader用于策略回测。如果你的电脑配置一般，可以先用更轻量的库组合，等熟悉了再升级到更完整的框架。

关于数据源的选择，国内A股最常用的免费数据接口是tushare和akshare。tushare需要注册并申请积分才能获取完整数据，但它的数据质量比较稳定，接口文档也比较清晰。akshare是纯免费的开源项目，数据覆盖面广，但接口稳定性稍差一些。我个人的建议是，新手先用akshare熟悉数据获取的基本流程，等对数据结构和接口调用有了一定了解之后，再切换到tushare获取更专业的数据服务。在安装这两个库时，需要注意它们的依赖版本，避免和已有的库产生冲突。建议在虚拟环境中单独测试每一个库，确保能够正常获取数据后再进行下一步。

# 安装量化交易所需的核心库（Windows PowerShell 或 Mac/Linux Terminal）
pip install pandas numpy akshare matplotlib backtrader

# 验证安装是否成功
python -c "import pandas; import numpy; import akshare; import matplotlib; import backtrader; print('All packages installed successfully!')"

# 如果遇到版本冲突，可以先升级pip
pip install --upgrade pip

除了Python环境之外，我还需要提醒大家准备好以下基础设施。第一是数据存储方案，建议使用SQLite这种轻量级数据库来存储历史数据，它比Excel文件更稳定，也更容易进行复杂查询。第二是回测结果可视化，建议用matplotlib或者plotly来绘制K线图和资金曲线，这会让你对策略的表现有更直观的认识。第三是交易信号推送，可以设置邮件或者微信通知，当策略发出交易信号时第一时间收到通知。这些工具虽然不是必须品，但能够大大提升你的量化交易体验。建议在正式编写策略之前，先花半天时间把这些辅助工具配置好。

02 数据获取：Python自动下载股票历史数据

用Python获取股票历史数据是量化策略的基础 图源受访者

数据是量化交易的基石，没有准确的数据，再好的策略也是空中楼阁。在这一节，我将教你如何用akshare获取A股股票的历史K线数据。akshare的优势在于接口简洁，数据种类丰富，不仅可以获取股票，还可以获取基金、期货、期权等多种资产的数据。唯一的缺点是数据更新速度略慢，但对于日线级别的量化策略来说，这个延迟完全可以接受。下面我们先从获取单只股票的历史数据开始，逐步构建起完整的数据获取体系。

import akshare as ak
import pandas as pd
import os

def download_stock_data(stock_code, start_date, end_date, save_path=None):
    """
    下载指定股票的历史数据
    stock_code: 股票代码，如"000001"代表平安银行，"600519"代表茅台
    start_date: 开始日期，格式"YYYYMMDD"
    end_date: 结束日期，格式"YYYYMMDD"
    """
    try:
        # 使用akshare获取股票日线数据
        df = ak.stock_zh_a_hist(
            symbol=stock_code,
            period="daily",
            start_date=start_date,
            end_date=end_date,
            adjust="qfq"  # 前复权数据，更适合技术分析
        )
        
        # 清理列名，去除空格
        df.columns = df.columns.str.strip()
        
        # 重命名列，便于后续处理
        df = df.rename(columns={
            '日期': 'date',
            '开盘': 'open',
            '收盘': 'close',
            '最高': 'high',
            '最低': 'low',
            '成交量': 'volume',
            '成交额': 'turnover',
            '振幅': 'amplitude',
            '涨跌幅': 'pct_change',
            '涨跌额': 'change',
            '换手率': 'turnover_rate'
        })
        
        # 将日期列转换为datetime格式
        df['date'] = pd.to_datetime(df['date'])
        
        # 按日期排序
        df = df.sort_values('date').reset_index(drop=True)
        
        print(f"成功获取 {stock_code} 的历史数据，共 {len(df)} 条记录")
        print(f"数据时间范围: {df['date'].min()} 至 {df['date'].max()}")
        
        # 如果提供了保存路径，则保存到本地
        if save_path:
            df.to_csv(save_path, index=False, encoding='utf-8-sig')
            print(f"数据已保存至: {save_path}")
        
        return df
        
    except Exception as e:
        print(f"获取数据失败: {str(e)}")
        return None

# 示例：获取平安银行2020年1月1日至2024年12月31日的日线数据
df = download_stock_data(
    stock_code="000001",
    start_date="20200101",
    end_date="20241231",
    save_path="stock_data/000001.csv"
)

# 查看数据前5行
print(df.head())

上面的代码实现了最基本的数据获取功能，但在实际应用中，你需要获取大量股票的数据来构建股票池。这时候就需要一个批量下载的函数来提高效率。我通常会把股票分为三类：沪深300成分股、中证500成分股和其他关注的股票，然后分批下载数据。为了避免请求过于频繁导致被封IP，我会在每次请求之间增加一个随机延迟。下载完成后的数据我会存储到本地的SQLite数据库中，这样后续的回测和分析都可以直接读取数据库，而不需要重复下载数据。

import sqlite3
import time
import random

def init_database(db_path="stock_data.db"):
    """初始化SQLite数据库"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS stock_daily (
            stock_code TEXT,
            date DATE,
            open REAL,
            close REAL,
            high REAL,
            low REAL,
            volume REAL,
            turnover REAL,
            pct_change REAL,
            turnover_rate REAL,
            PRIMARY KEY (stock_code, date)
        )
    """)
    conn.commit()
    return conn

def batch_download_and_save(stock_codes, start_date, end_date, db_path="stock_data.db"):
    """批量下载股票数据并存入数据库"""
    conn = init_database(db_path)
    
    for i, code in enumerate(stock_codes):
        print(f"[{i+1}/{len(stock_codes)}] 正在下载 {code} ...")
        df = download_stock_data(code, start_date, end_date)
        
        if df is not None and not df.empty:
            # 存入数据库
            df[['stock_code', 'date', 'open', 'close', 'high', 'low', 
                'volume', 'turnover', 'pct_change', 'turnover_rate']].to_sql(
                'stock_daily', conn, if_exists='append', index=False
            )
            print(f"  {code} 已存入数据库")
        
        # 随机等待1-3秒，避免请求过快
        time.sleep(random.uniform(1, 3))
    
    conn.close()
    print("批量下载完成！")

# 示例：批量下载多只股票
stock_list = ["000001", "000002", "600519", "600036", "000858"]
batch_download_and_save(stock_list, "20200101", "20241231")

03 策略实现：用Python实现双均线交叉策略

均线交叉策略是最经典的量化交易策略之一 图源受访者

均线交叉策略是技术分析中最经典的策略之一，它的原理很简单：当短期均线上穿长期均线时（金叉），产生买入信号；当短期均线下穿长期均线时（死叉），产生卖出信号。这个策略的逻辑符合趋势投资的核心理念——顺势而为。均线交叉策略的优点是逻辑清晰、易于实现、参数少；缺点是在震荡市中容易产生频繁的虚假信号，导致过度交易和亏损。我将通过代码逐步讲解如何用Python实现这个策略，并添加一些改进来降低虚假信号的干扰。

import pandas as pd
import numpy as np

class MovingAverageCrossStrategy:
    """双均线交叉策略"""
    
    def __init__(self, short_window=5, long_window=20):
        """
        初始化策略参数
        short_window: 短期均线周期（默认5日）
        long_window: 长期均线周期（默认20日）
        """
        self.short_window = short_window
        self.long_window = long_window
        
    def calculate_ma(self, df):
        """计算移动平均线"""
        df = df.copy()
        df['ma_short'] = df['close'].rolling(window=self.short_window).mean()
        df['ma_long'] = df['close'].rolling(window=self.long_window).mean()
        return df
    
    def generate_signals(self, df):
        """生成交易信号"""
        df = self.calculate_ma(df)
        
        # 初始化信号列：0表示无信号，1表示买入，-1表示卖出
        df['signal'] = 0
        
        # 当短期均线从下方穿越到上方时，产生买入信号
        df.loc[df['ma_short'] > df['ma_long'], 'signal'] = 1
        
        # 当短期均线从上方穿越到下方时，产生卖出信号
        df.loc[df['ma_short'] <= df['ma_long'], 'signal'] = -1
        
        # 只保留均线交叉的信号（即信号发生变化的那一天）
        df['position'] = df['signal'].diff()
        
        # 第一次交叉时也产生信号
        df.loc[df.index[0], 'position'] = df.loc[df.index[0], 'signal']
        
        return df
    
    def backtest(self, df, initial_capital=100000):
        """
        回测策略表现
        initial_capital: 初始资金
        """
        df = self.generate_signals(df)
        
        # 初始化资金和持仓
        cash = initial_capital
        position = 0  # 持仓股数
        shares = []  # 记录每日持仓
        
        for i, row in df.iterrows():
            # 买入信号
            if row['position'] == 2:  # 从无信号变为买入
                shares_to_buy = cash // row['close']
                if shares_to_buy > 0:
                    cash -= shares_to_buy * row['close']
                    position += shares_to_buy
                    print(f"买入日期: {row['date'].strftime('%Y-%m-%d')}, 买入价: {row['close']:.2f}, 买入数量: {shares_to_buy}")
            
            # 卖出信号
            elif row['position'] == -2 and position > 0:  # 从持仓变为卖出
                cash += position * row['close']
                print(f"卖出日期: {row['date'].strftime('%Y-%m-%d')}, 卖出价: {row['close']:.2f}, 卖出数量: {position}")
                position = 0
            
            # 计算当日总资产
            total_value = cash + position * row['close']
            shares.append({
                'date': row['date'],
                'cash': cash,
                'position': position,
                'total_value': total_value
            })
        
        # 生成回测报告
        result_df = pd.DataFrame(shares)
        result_df['return'] = result_df['total_value'].pct_change()
        
        # 计算关键指标
        total_return = (result_df['total_value'].iloc[-1] - initial_capital) / initial_capital * 100
        annual_return = total_return / (len(result_df) / 252)  # 假设一年252个交易日
        
        # 最大回撤
        result_df['cummax'] = result_df['total_value'].cummax()
        result_df['drawdown'] = (result_df['total_value'] - result_df['cummax']) / result_df['cummax']
        max_drawdown = result_df['drawdown'].min() * 100
        
        print(f"\n===== 回测结果 =====")
        print(f"总收益率: {total_return:.2f}%")
        print(f"年化收益率: {annual_return:.2f}%")
        print(f"最大回撤: {max_drawdown:.2f}%")
        
        return result_df

# 使用示例
strategy = MovingAverageCrossStrategy(short_window=5, long_window=20)
result = strategy.backtest(df, initial_capital=100000)

上面的代码实现了一个最基本的双均线交叉策略，但它还有很多可以优化的地方。首先，当前的策略没有加入止损机制，如果在单边行情中出现大幅亏损，策略无法自动止损；其次，策略没有考虑交易成本，频繁交易会大大降低实际收益；再次，策略没有做仓位管理，每次都是全仓买卖，风险过于集中。下面我将对这些问题进行逐一优化，让策略更加贴近实盘环境。

class ImprovedMAcrossStrategy:
    """改进版双均线交叉策略：加入止损、分批建仓和仓位管理"""
    
    def __init__(self, short_window=5, long_window=20, 
                 stop_loss_pct=0.08, take_profit_pct=0.15,
                 position_ratio=0.3):
        self.short_window = short_window
        self.long_window = long_window
        self.stop_loss_pct = stop_loss_pct  # 止损比例8%
        self.take_profit_pct = take_profit_pct  # 止盈比例15%
        self.position_ratio = position_ratio  # 每次建仓比例30%
        
    def backtest_improved(self, df, initial_capital=100000):
        df = df.copy()
        df['ma_short'] = df['close'].rolling(window=self.short_window).mean()
        df['ma_long'] = df['close'].rolling(window=self.long_window).mean()
        df['signal'] = 0
        
        # 计算均线交叉信号
        df.loc[df['ma_short'] > df['ma_long'], 'signal'] = 1
        df.loc[df['ma_short'] <= df['ma_long'], 'signal'] = -1
        df['position'] = df['signal'].diff()
        df.loc[df.index[0], 'position'] = df.loc[df.index[0], 'signal']
        
        cash = initial_capital
        position = 0
        buy_price = 0
        shares = []
        
        for i, row in df.iterrows():
            current_price = row['close']
            
            # 如果持仓中，检查止损和止盈
            if position > 0:
                price_change = (current_price - buy_price) / buy_price
                
                # 止损
                if price_change <= -self.stop_loss_pct:
                    cash += position * current_price
                    print(f"[止损] 日期: {row['date'].strftime('%Y-%m-%d')}, 价格: {current_price:.2f}, 亏损: {price_change*100:.2f}%")
                    position = 0
                    buy_price = 0
                
                # 止盈
                elif price_change >= self.take_profit_pct:
                    cash += position * current_price
                    print(f"[止盈] 日期: {row['date'].strftime('%Y-%m-%d')}, 价格: {current_price:.2f}, 盈利: {price_change*100:.2f}%")
                    position = 0
                    buy_price = 0
            
            # 金叉信号：买入
            if row['position'] == 2 and position == 0:
                use_capital = cash * self.position_ratio
                shares_to_buy = int(use_capital / current_price)
                if shares_to_buy > 0:
                    cost = shares_to_buy * current_price
                    commission = cost * 0.0003  # 佣金万三
                    total_cost = cost + commission
                    if total_cost <= cash:
                        cash -= total_cost
                        position += shares_to_buy
                        buy_price = current_price
                        print(f"[买入] 日期: {row['date'].strftime('%Y-%m-%d')}, 价格: {current_price:.2f}, 数量: {shares_to_buy}")
            
            # 死叉信号：卖出
            elif row['position'] == -2 and position > 0:
                commission = position * current_price * 0.0003
                cash += position * current_price - commission
                print(f"[卖出] 日期: {row['date'].strftime('%Y-%m-%d')}, 价格: {current_price:.2f}, 数量: {position}")
                position = 0
                buy_price = 0
            
            shares.append({
                'date': row['date'],
                'cash': cash,
                'position': position,
                'total_value': cash + position * current_price
            })
        
        result_df = pd.DataFrame(shares)
        result_df['return'] = result_df['total_value'].pct_change()
        
        total_return = (result_df['total_value'].iloc[-1] - initial_capital) / initial_capital * 100
        annual_return = total_return / (len(result_df) / 252)
        
        result_df['cummax'] = result_df['total_value'].cummax()
        result_df['drawdown'] = (result_df['total_value'] - result_df['cummax']) / result_df['cummax']
        max_drawdown = result_df['drawdown'].min() * 100
        
        print(f"\n===== 改进策略回测结果 =====")
        print(f"总收益率: {total_return:.2f}%")
        print(f"年化收益率: {annual_return:.2f}%")
        print(f"最大回撤: {max_drawdown:.2f}%")
        
        return result_df

# 运行改进版策略
improved_strategy = ImprovedMAcrossStrategy(
    short_window=5, 
    long_window=20,
    stop_loss_pct=0.08,
    take_profit_pct=0.15,
    position_ratio=0.3
)
result = improved_strategy.backtest_improved(df, initial_capital=100000)

04 回测优化：参数优化与风险控制

回测优化是提升量化策略表现的关键步骤 图源受访者

回测是量化交易中不可或缺的环节，它帮助我们验证策略在过去市场中的表现，为实盘交易提供参考依据。但回测也是最容易出问题的地方，很多新手做的回测实际上是"过度拟合"的结果——策略在历史数据上表现很好，但在新数据上却一塌糊涂。这一节，我将讲解如何正确进行回测优化，包括参数遍历、样本内外测试、夏普比率分析等关键环节。掌握这些方法，能够帮助你避免过度拟合的陷阱，让策略在实盘中保持稳健。

from itertools import product
import numpy as np

def parameter_optimization(df, initial_capital=100000):
    """
    参数优化：遍历不同的均线周期组合
    找到最优的参数配置
    """
    # 定义参数范围
    short_windows = [3, 5, 7, 10, 15]
    long_windows = [20, 30, 40, 50, 60]
    
    results = []
    
    print("开始参数优化...")
    for short_w, long_w in product(short_windows, long_windows):
        # 短期均线必须小于长期均线
        if short_w >= long_w:
            continue
        
        strategy = ImprovedMAcrossStrategy(
            short_window=short_w,
            long_window=long_w,
            stop_loss_pct=0.08,
            take_profit_pct=0.15,
            position_ratio=0.3
        )
        
        result_df = strategy.backtest_improved(df.copy(), initial_capital)
        
        # 计算关键指标
        total_return = (result_df['total_value'].iloc[-1] - initial_capital) / initial_capital * 100
        annual_return = total_return / (len(result_df) / 252)
        
        result_df['cummax'] = result_df['total_value'].cummax()
        result_df['drawdown'] = (result_df['total_value'] - result_df['cummax']) / result_df['cummax']
        max_drawdown = result_df['drawdown'].min() * 100
        
        # 计算夏普比率
        returns = result_df['return'].dropna()
        sharpe_ratio = (annual_return / 100) / returns.std() * np.sqrt(252) if returns.std() > 0 else 0
        
        results.append({
            'short_window': short_w,
            'long_window': long_w,
            'total_return': total_return,
            'annual_return': annual_return,
            'max_drawdown': max_drawdown,
            'sharpe_ratio': sharpe_ratio
        })
        
        print(f"参数 (MA{short_w}, MA{long_w}): 收益率 {total_return:.2f}%, 最大回撤 {max_drawdown:.2f}%, 夏普比率 {sharpe_ratio:.2f}")
    
    # 转为DataFrame并排序
    results_df = pd.DataFrame(results)
    results_df = results_df.sort_values('sharpe_ratio', ascending=False)
    
    print("\n===== 最优参数组合 =====")
    print(results_df.head(10))
    
    return results_df

# 运行参数优化
optimization_results = parameter_optimization(df)

参数优化完成后，我们需要对结果进行严格的筛选，不能简单地选择收益率最高的参数组合。这里有几个重要的原则需要遵守。第一是夏普比率优先原则：收益率高但风险也高的策略，不如收益率适中但风险可控的策略。夏普比率（Sharpe Ratio）是衡量策略性价比的核心指标，它表示每承担一单位风险所获得的超额收益。一般来说，夏普比率大于1的策略是值得考虑的，大于2的策略是优秀的。第二是最大回撤控制原则：即使收益率很高，如果最大回撤超过30%，策略的风险也是不可接受的。建议选择最大回撤在15%以内的策略，这样才能在极端市场环境中存活下来。

def walk_forward_validation(df, short_window, long_window, 
                           train_days=252, test_days=63, step=21):
    """
    walk-forward验证：
    用历史的252天训练，然后用接下来的63天测试
    逐步滚动，验证策略的稳定性
    """
    results = []
    start_idx = train_days
    end_idx = start_idx + test_days
    
    while end_idx <= len(df):
        # 分割训练集和测试集
        train_df = df.iloc[start_idx - train_days:start_idx].copy()
        test_df = df.iloc[start_idx:end_idx].copy()
        
        # 在训练集上优化参数
        train_strategy = ImprovedMAcrossStrategy(
            short_window=short_window,
            long_window=long_window
        )
        train_result = train_strategy.backtest_improved(train_df)
        
        # 在测试集上验证
        test_strategy = ImprovedMAcrossStrategy(
            short_window=short_window,
            long_window=long_window
        )
        test_result = test_strategy.backtest_improved(test_df)
        
        train_return = (train_result['total_value'].iloc[-1] - train_result['total_value'].iloc[0]) / train_result['total_value'].iloc[0] * 100
        test_return = (test_result['total_value'].iloc[-1] - test_result['total_value'].iloc[0]) / test_result['total_value'].iloc[0] * 100
        
        results.append({
            'train_period': f"{train_df['date'].iloc[0].strftime('%Y-%m-%d')} ~ {train_df['date'].iloc[-1].strftime('%Y-%m-%d')}",
            'test_period': f"{test_df['date'].iloc[0].strftime('%Y-%m-%d')} ~ {test_df['date'].iloc[-1].strftime('%Y-%m-%d')}",
            'train_return': train_return,
            'test_return': test_return,
            'out_of_sample': test_return / train_return if train_return != 0 else 0
        })
        
        print(f"训练期: {results[-1]['train_period']}, 收益率: {train_return:.2f}%")
        print(f"测试期: {results[-1]['test_period']}, 收益率: {test_return:.2f}%, OOS比率: {results[-1]['out_of_sample']:.2f}")
        
        # 移动到下一个窗口
        start_idx += step
        end_idx += step
    
    results_df = pd.DataFrame(results)
    print("\n===== Walk-Forward验证结果 =====")
    print(f"平均样本内收益率: {results_df['train_return'].mean():.2f}%")
    print(f"平均样本外收益率: {results_df['test_return'].mean():.2f}%")
    print(f"平均OOS比率: {results_df['out_of_sample'].mean():.2f}")
    
    return results_df

# 运行walk-forward验证
wf_results = walk_forward_validation(df, short_window=5, long_window=20)

05 实盘模拟：从回测到实盘的第一步

实盘模拟是验证策略有效性的关键过渡步骤 图源受访者

完成回测和优化之后，下一步就是进入实盘模拟阶段。实盘模拟的意义在于，在真实的市场环境中测试策略的表现，同时不承受真实的资金损失。这个阶段需要解决的问题包括：如何将策略信号转化为实际的交易指令，如何处理交易滑点和延迟，如何监控策略表现并及时发现异常。在这一节，我将介绍如何构建一个完整的实盘模拟框架，帮助你平滑过渡到真实交易。

class PaperTradingSimulator:
    """
    实盘模拟交易器
    模拟真实的交易流程，包括信号生成、订单执行、持仓管理等
    """
    
    def __init__(self, initial_capital=100000, commission_rate=0.0003):
        self.initial_capital = initial_capital
        self.commission_rate = commission_rate
        self.cash = initial_capital
        self.position = 0
        self.buy_price = 0
        self.trade_log = []
        
    def execute_buy(self, date, price, shares):
        """执行买入"""
        cost = shares * price
        commission = cost * self.commission_rate
        total_cost = cost + commission
        
        if total_cost <= self.cash:
            self.cash -= total_cost
            self.position += shares
            self.buy_price = price
            self.trade_log.append({
                'date': date,
                'action': 'BUY',
                'price': price,
                'shares': shares,
                'commission': commission,
                'cash': self.cash,
                'position': self.position
            })
            return True
        return False
    
    def execute_sell(self, date, price, shares=None):
        """执行卖出，shares=None表示全部卖出"""
        if self.position == 0:
            return False
        
        if shares is None:
            shares = self.position
        
        shares = min(shares, self.position)
        revenue = shares * price
        commission = revenue * self.commission_rate
        net_revenue = revenue - commission
        
        self.cash += net_revenue
        self.position -= shares
        self.trade_log.append({
            'date': date,
            'action': 'SELL',
            'price': price,
            'shares': shares,
            'commission': commission,
            'cash': self.cash,
            'position': self.position
        })
        
        if self.position == 0:
            self.buy_price = 0
        
        return True
    
    def check_stop_loss(self, current_price, stop_loss_pct=0.08):
        """检查是否触发止损"""
        if self.position == 0 or self.buy_price == 0:
            return False
        return (current_price - self.buy_price) / self.buy_price <= -stop_loss_pct
    
    def check_take_profit(self, current_price, take_profit_pct=0.15):
        """检查是否触发止盈"""
        if self.position == 0 or self.buy_price == 0:
            return False
        return (current_price - self.buy_price) / self.buy_price >= take_profit_pct
    
    def get_total_value(self, current_price):
        """计算当前总资产"""
        return self.cash + self.position * current_price
    
    def get_returns(self):
        """计算当前收益率"""
        return (self.get_total_value(self.buy_price if self.position > 0 else 0) - self.initial_capital) / self.initial_capital * 100

# 模拟实盘交易
simulator = PaperTradingSimulator(initial_capital=100000)

# 假设有一个策略信号DataFrame
for i, row in df.iterrows():
    # 假设这是一个实盘信号
    if row['position'] == 2 and simulator.position == 0:  # 买入信号
        simulator.execute_buy(row['date'], row['close'], 100)
        print(f"模拟买入: {row['date'].strftime('%Y-%m-%d')}, 价格: {row['close']:.2f}")
    
    elif row['position'] == -2 and simulator.position > 0:  # 卖出信号
        simulator.execute_sell(row['date'], row['close'])
        print(f"模拟卖出: {row['date'].strftime('%Y-%m-%d')}, 价格: {row['close']:.2f}")
    
    # 检查止损止盈
    elif simulator.position > 0:
        if simulator.check_stop_loss(row['close']):
            simulator.execute_sell(row['date'], row['close'])
            print(f"止损卖出: {row['date'].strftime('%Y-%m-%d')}, 价格: {row['close']:.2f}")
        elif simulator.check_take_profit(row['close']):
            simulator.execute_sell(row['date'], row['close'])
            print(f"止盈卖出: {row['date'].strftime('%Y-%m-%d')}, 价格: {row['close']:.2f}")

print(f"\n===== 实盘模拟结果 =====")
print(f"最终资金: {simulator.cash:.2f}")
print(f"持仓: {simulator.position}")
print(f"交易次数: {len(simulator.trade_log)}")

实盘模拟阶段需要注意几个关键问题。第一是信号延迟问题：回测中我们使用的是收盘价，但实盘中你只能在收盘后才知道收盘价，所以实际买入价格会比回测中的价格差一些。这个差异叫做"收盘价偏差"，对于短线策略影响很大。解决方案是使用更短周期的数据（如分钟线），或者在回测中模拟更保守的成交价格。第二是滑点问题：实盘成交价格往往比下单价格差一些，特别是对于流动性较差的股票。建议在回测中把滑点设置得比实际预期更高，比如设置为0.5%到1%，这样可以让回测结果更加保守。第三是交易执行问题：实盘中可能会遇到下单失败、成交数量不足等情况，需要有完善的错误处理机制。

免责声明：本文仅供学习交流，不构成任何投资建议。量化交易存在风险，请谨慎评估后决策。