1、量化交易入门

约 3253 字大约 11 分钟

2026-05-30

量化交易开发流程

总体开发流程

数据获取
数据清洗
策略编写（历史数据回测）
策略回测（历史数据回测）
策略优化（历史数据回测）
模拟盘交易（未来数据模拟）
实盘交易

开发流程详解

数据获取

行情数据
宏观数据
财务数据
舆情数据

数据获取方法：

网站下载
客户端
三方api
爬虫

数据清洗

垃圾数据清除
空值填充
格式转换
数据对齐

数据清洗常用类库：

pandas
numpy

策略编写

信号捕捉：交易信号的捕捉，交易买卖；
交易
1. 建仓：买
2. 平仓：卖

策略回测

回测参数设置
策略实例化
历史数据载入
回测执行
计算盈亏
计算统计指标
生成回测报告

量化策略优化

重视交易费
重视风险、重视退出
优化无止境

模拟盘交易

过去的表现并不表示未来的结果
保持模拟盘交易至少半年以上
模拟盘稳定收益100%以上在考虑实盘交易

实盘交易

做好第一年会输的准备
不急于扩大投资，增加杠杆

量化交易策略分类

量化交易策略分类体系

量化交易
- 按照交易产品分类
  - 股票策略（股票）
  - 期权策略（期权）
  - CTA策略（期货）
  - FOF策略（FOF）
- 盈利模式分类
  - 单边多空策略（做多或做空）
  - 套利策略：稳定的无风险的收益率
  - 对冲策略：
- 策略信号【数学模型】分类
  - 多因子策略
  - 均值回归策略
  - 动量效益策略
  - 二八轮动策略
  - 海龟策略
  - 机器学习策略

交易产品分类

交易产品分类有以下四种:

股票策略（股票,主要通过股价的波动）
期权策略（期权主要通过期权的价差获利）
CTA策略（期货通过价格走势的背离获利）
FOF策略（FOF 分担风险）

交易产品相关量化策略盈利模式

股票策略盈利模式

期权和期货

期权和期货对比:

风险现货最小，期权风险最大；

期权策略盈利模式

CTA策略盈利模式

FOF策略

根据盈利模式分类

盈利模式分类介绍

单边多空策略，单即看跌，多即看涨
套利策略，支持多种标的，期权期货，债券，股票
对冲策略

各量化策略定义及盈利模式

单边多空策略

套利策略

对冲策略

什么是对冲:

对冲策略：

根据策略信号分类

策略信号分类介绍

多因子
交易模型
机器学习

各量化策略定义及盈利模式

策略信号:

多因子策略

交易模型

机器学习

常见金融标的的投资风险与收益

股票分类

蓝筹股

白马股

成长股

周期股

概念股

股票行业分类

中证行业分类:

申万行业分类:

行业分类作用

影响股价因素

影响股价的因素

6大因素详解

经济因素:

政治因素:

行业因素:

企业自身因素:

市场因素:

心理因素:

股票交易基础知识

K线

财务知识

选股

选股定义

基本面选股

估值方法

择时

择时定义

技术分析

6种常用技术分析工具

K线形态:

成交量:

均线:

布林带:

MACD:

KDJ:

量化交易平台

国内常见量化交易平台

什么是量化交易平台?

聚宽量化交易平台

聚宽量化交易平台在线使用

聚宽量化交易平台地址

Numpy

数据样本说明

股价常用指标实现

简易量化交易策略实现

策略框架

实用策略示例

策略编写:

def initialize(context):
    g.security = '000001.XSHE'
    run_daily(market_open,time='every_bar')
    
    
def market_open(context):
    security = g.security
    close_data = attribute_history(security, 5, '1d', ['close'])
    MA5 = close_data['close'].mean()
    current_price = close_data['close'][-1]
    
    cash = context.portfolio.available_cash
    
    
    if current_price > 1.01*MA5:
        order_value(security,cash)
        log.info("买入 %s" % (security))
    
    elif current_price < MA5 and context.portfolio.positions[security].closeable_amount >0:
        order_target(security,0)
        log.info("卖出 %s" % (security))
        
    record(stock_price=current_price)

策略函数设置

常用策略设置函数

定时函数

什么是定时函数?

定时函数参数:

交易函数

1.交易函数

（1）交易数量

order(security, amount, style=None, side='long', pindex=0)

security: 股票代码

amount: 交易数量（负数表示卖出）

style: 下单类型

side: short空（一般不允许）/ long多

pindex: 仓位号，默认为0

示例：

按限单价30买入100股

order('600000.XSHG', 100, LimitOrderStyle(30.0))

（2）股票价值

order(security, value, style=None, side='long', pindex=0)

security: 股票代码

value: 股票价值（负数表示卖出）

style: 下单类型

side: short空（一般不允许）/ long多

pindex: 仓位号，默认为0

示例：

按卖出价值为5000元的股票

order('600000.XSHG', -5000)

（3）目标数量

order_target(security, amount, style=None, side='long', pindex=0, close_today=False)

security: 股票代码

amount: 交易数量（负数表示卖出）

style: 下单类型

side: short空（一般不允许）/ long多

pindex: 仓位号，默认为0

示例：

买入平安银行所有股票到100股

order_target('000001.XSHE', 100)

（4）查询成交订单

get_orders(order id=None,security=None,status=None)

示例：

get_orders(order_id='123)订单id查询订单号为"123"的订单

get_orders(security='000001.XSHE') 查询所有标的为000001.XSHE的订单

（5）未完成订单

get_open_orders()

示例：

在每天交易结束之后获取当天所有的未完成订单

def after_market_close(context):

orders = get_open_orders(）

for order in orders:

log.info(_order.order_id)

（6）撤单函数

cancel_order(order)

示例：

在每天交易结束之后对未完成订单进行撤单

def after_market_close(context):

orders = get_open_orders()

for _order in orders.values():

cancel_order(order)

（7）账号出入金

inout_cash(cash, pindex=0)

cash: 浮点数，负数表示出金

pindex: 仓位号，默认为0

示例：

向账户增加10000元:

inout_cash(10000, pindex=0)

代码实战

def initialize(context):
    # 设定业绩标的为“沪深300”
    set_benchmark("000300.XSHG")
    g.security = '000001.XSHE'
    run_daily(market_open, time='9:30')
    run_daily(after_market_close, time='15:30')

def market_open(context):
    # 向账户增加10000元
    inout_cash(10000, pindex=0)
    # 查询可以资金
    log.info("""账户可用资金：{}""".format(context.portfolio.subportfolios[0].available_cash))
    
    # 获取账户当前现金
    cash = context.portfolio.available_cash
    
    #如果没有持仓
    if g.security not in context.portfolio.positions:
        #下单1000股
        order(g.security, 1000)
    else:
        #卖出500股
        order(g.security, -500)
        
        
def after_market_close(context):
    # 获取当天所有未完成的订单
    orders = get_open_orders()
    for _order in orders:
        log.info("""未完成订单：{}""".format(_order))
    # 对未完成订单进行撤单
    for _order in orders:
        cancel_order(_order)

交易对象

1.Order对象

Order对象：订单处理流程

订单创建 ->订单检查 -> 报单 -> 确认委托 -> 撮合

commission: 交易费用(佣金、税费等)

is_buy: bool值,买还是卖

status: 状态,一个OrderStatus值

price: 平均成交价格，已经成交的股票的平均成交价格

2.Trade对象

Trade对象：订单成交相关信息

time：交易时间，[datetime.datetime]对象
security：标的代码
amount：交易数量
price：交易价格
trade_id：交易记录id
order_id：对应的订单id

代码案例

# 初始化函数：回测开始前只执行一次
def initialize(context):
    # 设置业绩基准：沪深300指数
    set_benchmark("000300.XSHG")
    # 指定交易股票：平安银行(000001)
    g.security = '000001.XSHE'
    
    # 每天开盘自动执行交易函数
    run_daily(market_open, time='open')
    # 每天15:30收盘后执行日志记录函数
    run_daily(after_market_close, time='15:30')

# 开盘交易逻辑：每天开盘自动运行
def market_open(context):
    # 如果没有持仓 → 买入1000股
    if g.security not in context.portfolio.positions:
        order(g.security, 1000)
    # 如果有持仓 → 卖出800股
    else:
        order(g.security, -800)

# 收盘后函数：每天15:30自动运行
def after_market_close(context):
    print("闭市后，开始查询成交记录...")
    
    # 获取当天所有成交记录（字典格式）
    trades = get_trades()
    
    # 遍历所有成交记录
    for _trade in trades.values():
        # 打印成交详情
        print("成交记录:{}".format(_trade))
        print("成交时间:{}".format(_trade.time))
        print("对应的订单id:{}".format(_trade.order_id))


"""
def market_open(context):
    #如果没有持仓
    if g.security not in context.portfolio.positions:
        orders = order(g.security, 100)
        print(orders)
        
        if orders is None:
            print("创建订单失败...")
        else:
            print('''交易费用：{}'''.format(orders.commission))
            print('''是否买单：{}'''.format(orders.is_buy))
            print('''订单状态：{}'''.format(orders.status))
            print('''订单评价成交价格：{}'''.format(orders.price))
    else:
        #卖出500股
        order(g.security, -800)
"""

策略信息

1.Context对象

Context对象：策略信息总览，包含账户、时间等信息
subportfolios：当前单个操作仓位的资金、标的信息，是一个SubPortfolio的数组
portfolio：账户信息，即subportfolios 的汇总信息，Portfolio对象，单个操作仓位时，portfolio指向 subportfolios[0]
current_dt：当前单位时间的开始时间，[datetime.datetime]对象
previous date：datetime前一个交易日，[datetime.date]对象,注意，这是一个日期，是 date,而不是datetime
universe：查询set_universe()设定的股票池，比如['000001.XSHE','600000.XSHG']

2.Position对象

Position对象：输出持有的标的的信息

security：标的代码
price：最新行情价格
total_amount：总仓位，不包括挂单冻结仓位
init_time：建仓时间,格式为datetime.datetime

代码案例

def initialize(context):
    g.security = "000001.XSHE"
    
def handle_data(context, data):
    #如果没有持仓
    if g.security not in context.portfolio.positions:
        #下单1000股
        order(g.security, 1000)
    else:
        #卖出800股
        order(g.security, -800)
    print(type(context.portfolio.long_positions))
    long_positions_dict = context.portfolio.long_positions
    for position in list(long_positions_dict.values()):
        print(
            """"标的:{}，总仓位:{}，标的价值:{}，建仓时间:{}"""
            .format(position.security,
                position.total_amount,
                position.value,
                position.init_time)
        )
    

def handle_data(context, data):
    #context.portfolio变为整数1
    context.portfolio = 1
    log.info("context.portfolio")
    log.info(context.portfolio)
    
    #恢复系统变量
    del context.portfolio
    
    #context.portfolio将变成用户账户信息
    log.info("context.portfolio.total_value")
    log.info(context.portfolio.total_value)
    
    # 输出账户总资产
    log.info("context.portfolio.total_value")
    log.info(context.portfolio.total_value)
    
    # 输出持仓金额
    log.info("context.portfolio.positions_value")
    log.info(context.portfolio.positions_value)
    
    # 输出今日日期
    log.info("context.current_dt.day")
    log.info(context.current_dt.day)
    
    # 输出总权益的累计收益
    log.info("context.portfolio.returns")
    log.info(context.portfolio.returns)
    
    # 获取仓位subportfolios[0]的可用资金
    log.info("context.subportfolios[0].available_cash")
    log.info(context.subportfolios[0].available_cash)

账户信息

1.Portfolio对象

Portfolio对象：总账户信息

long_positions：

多单的仓位,一个dict, key是证券代码, value 是[Position]对象

short_positions：

空单的仓位,一个dict, key是证券代码, value 是[Position]对象

total_value：

总的权益，包括现金，保证金(期货)或者仓位(股票)的总价值，可用来计算收益

returns：

总权益的累计收益;(当前总资产+今日出入金-昨日总资产)/昨日总资产

starting_cash：

初始资金,现在等于inout_cash

positions_value：

持仓价值

2.SubPortfolio对象

SubPortfolio对象：子账户信息

inout_cash：

累计出入金,如初始资金1000,后来转移出去100,则这个值是1000 -100

available_cash：

可用资金,可用来购买证券的资金

transferable_cash：

可取资金，即可以提现的资金,不包括今日卖出证券所得资金

locked_cash：

挂单锁住资金

type：

账户所属类型

代码案例

def initialize(context):
    g.security = "000001.XSHE"
    
def handle_data(context, data):
    #如果没有持仓
    if g.security not in context.portfolio.positions:
        #下单1000股
        order(g.security, 1000)
    else:
        #卖出800股
        order(g.security, -800)
    
    print("""累计出入金:{}""".format(context.subportfolios[0].inout_cash))
    print("""可以资金:{}""".format(context.subportfolios[0].available_cash))
    print("""可取资金:{}""".format(context.subportfolios[0].transferable_cash))
    print("""挂单锁住资金:{}""".format(context.subportfolios[0].locked_cash))
    print("""账户所属类型:{}""".format(context.subportfolios[0].type))

'''
    print("""多单的仓位:{}""".format(context.portfolio.long_positions))
    print("""空单的仓位:{}""".format(context.portfolio.short_positions))
    print("""总权益:{}""".format(context.portfolio.total_value))
    print("""总权益的累计收益率:{}""".format(context.portfolio.returns))
    print("""初始资金:{}""".format(context.portfolio.starting_cash))
    print("""持仓价值:{}""".format(context.portfolio.positions_value))
'''

贡献者

rzfgentlevip