Python 量化交易
学习手册

python

print("你好，QMT！")
print("策略开始运行")
print("=" * 80)

# 这是单行注释，# 后面的内容不会被执行

"""
这是多行注释（文档字符串）
可以写多行说明
常用于函数开头的功能描述
"""

# 把 100 赋值给变量 stock_num
stock_num = 100

# 把文字赋值给变量 name
name = "贵州茅台"

# 使用变量
print(name, stock_num)

# 整数
a = 10
b = 3

# 浮点数
price = 25.68
weight = 1.0

# 四则运算
print(a + b)    # 13    加法
print(a - b)    # 7     减法
print(a * b)    # 30    乘法
print(a / b)    # 3.333...  除法（结果永远是浮点数）
print(a // b)   # 3     整除（只保留整数部分）
print(a % b)    # 1     取余数

# 其他常用运算
print(abs(-5))        # 5       绝对值
print(round(3.1416, 2))  # 3.14  四舍五入
print(max(10, 20, 5))    # 20    最大值
print(min(10, 20, 5))    # 5     最小值

# and：两个条件都为 True 才是 True
price = 15.5
pre_close = 14.0
print(price > 0 and pre_close > 0)   # True

# or：只要有一个为 True 就是 True
code = "300001.SZ"
print(code.startswith("68") or code.startswith("30"))  # True

# not：取反
is_st = False
print(not is_st)   # True

stock = "000001.SZ"
name = '平安银行'

# 字符串拼接
full = stock + " " + name
print(full)   # 000001.SZ 平安银行

# 字符串重复（策略中的分隔线就是这么做的）
print("=" * 40)   # ========================================

# f-string：在字符串中嵌入变量（Python 3.6+）
price = 15.68
print(f"{stock} 当前价格: {price}")   # 000001.SZ 当前价格: 15.68
print(f"{stock} 涨停价: {price * 1.1:.2f}")  # 000001.SZ 涨停价: 17.25

# 股票代码格式："300001.SZ" 或 "600519.SH"
stock = "300001.SZ"

# 提取纯数字部分
code_part = stock.split('.')[0]    # "300001"

# 判断板块
if code_part.startswith("68"):     # 科创板
    print("科创板，涨跌幅20%")
elif code_part.startswith("30"):   # 创业板
    print("创业板，涨跌幅20%")
elif code_part.startswith("4") or code_part.startswith("8"):  # 北交所
    print("北交所，过滤掉")
else:                               # 主板
    print("主板，涨跌幅10%")

# 检查股票名称中是否包含 ST
stock_name = "*ST大控"
if "ST" in stock_name or "退" in stock_name:
    print("ST或退市股，过滤掉")

date_str = "20260718"

year = date_str[0:4]      # "2026"
month = date_str[4:6]     # "07"
day = date_str[6:8]       # "18"

# 负数索引：从末尾开始
code = "600519.SH"
exchange = code[-2:]      # "SH"  （最后两个字符）
code_only = code[:-3]     # "600519" （去掉最后三个字符）

# 创建列表
stocks = ["600519.SH", "000858.SZ", "300750.SZ"]
numbers = [10, 20, 30, 40, 50]
mixed = ["茅台", 1800.5, True]   # 可以混合类型

# 通过索引访问（从 0 开始）
print(stocks[0])    # "600519.SH"  第一个
print(stocks[-1])   # "300750.SZ"  最后一个

# 获取列表长度
print(len(stocks))  # 3

holdings = ["600519.SH", "000858.SZ", "300750.SZ"]

# 添加元素
holdings.append("002475.SZ")       # 末尾添加
# holdings: ["600519.SH", "000858.SZ", "300750.SZ", "002475.SZ"]

# 删除元素
holdings.remove("000858.SZ")       # 按值删除（只删第一个匹配的）
holdings.pop(0)                    # 按索引删除并返回
# holdings: ["300750.SZ", "002475.SZ"]

# 检查是否包含
if "600519.SH" in holdings:
    print("已持有茅台")

# 按价格升序排序（小到大）
stock_prices = [("600519.SH", 1800.5), ("000001.SZ", 15.68), ("300750.SZ", 220.0)]
stock_prices.sort(key=lambda x: x[1])
# 结果: [("000001.SZ", 15.68), ("300750.SZ", 220.0), ("600519.SH", 1800.5)]

# 按市值升序（小到大）
stock_cap_list = [("A", 50e8), ("B", 20e8), ("C", 100e8)]
stock_cap_list.sort(key=lambda x: x[1], reverse=False)
# 结果: [("B", 20e8), ("A", 50e8), ("C", 100e8)]

# reverse=True 则降序（大到小）
stock_cap_list.sort(key=lambda x: x[1], reverse=True)

nums = [10, 20, 30, 40, 50, 60, 70]

nums[0:3]     # [10, 20, 30]   前三个
nums[-3:]     # [50, 60, 70]   最后三个
nums[2:5]     # [30, 40, 50]   中间一段
nums[:3]      # [10, 20, 30]   从头到第3个
nums[3:]      # [40, 50, 60, 70]  第4个到末尾

# 基本语法：[表达式 for 变量 in 列表 if 条件]

stocks = ["600519.SH", "000858.SZ", "300750.SZ", "002475.SZ"]

# 筛选创业板股票（30开头）
kcb = [s for s in stocks if s.startswith("30")]
# ["300750.SZ"]

# 从元组列表中提取代码
pairs = [("600519.SH", 1800.5), ("000001.SZ", 15.68)]
codes = [code for code, price in pairs]
# ["600519.SH", "000001.SZ"]

# 筛选增长率大于0的股票
positive = [stock for stock, growth in revenue_growth_dict.items() if growth > 0]

# 列表转字典
pairs = [("A", 100), ("B", 200)]
d = dict(pairs)
# d = {"A": 100, "B": 200}

# 列表转字符串
codes = ["600519", "000858", "300750"]
result = ",".join(codes)
# "600519,000858,300750"

# 创建字典
target_assets = {
    "600519.SH": 0.1,
    "000858.SZ": 0.1,
    "300750.SZ": 0.1
}

# 访问值
print(target_assets["600519.SH"])   # 0.1

# 添加/修改
target_assets["002475.SZ"] = 0.1    # 添加新键值对
target_assets["600519.SH"] = 0.2    # 修改已有值

# 删除
del target_assets["000858.SZ"]

# 安全访问（键不存在时不报错，返回默认值）
price = target_assets.get("999999.SH", 0)   # 0（键不存在，返回默认值）

holdings = {"600519.SH": 5, "000858.SZ": 10, "300750.SZ": 3}

# 遍历键和值
for stock, hand in holdings.items():
    print(f"{stock}: {hand}手")
# 输出：
# 600519.SH: 5手
# 000858.SZ: 10手
# 300750.SZ: 3手

# 只遍历键
for stock in holdings.keys():
    print(stock)

# 只遍历值
for hand in holdings.values():
    print(hand)

# 检查键是否存在
if "600519.SH" in holdings:
    print("持有茅台")

# 获取字典长度
print(len(holdings))   # 3

# QMT 的 get_market_data_ex 返回的数据结构类似这样：
hist_data = {
    "600519.SH": {
        "close": [1800.0, 1815.5, 1830.2],
        "high":  [1820.0, 1830.0, 1845.0],
        "low":   [1790.0, 1800.0, 1815.0]
    }
}

# 取某只股票的收盘价数据
close_data = hist_data["600519.SH"]["close"]
# [1800.0, 1815.5, 1830.2]

# 判断是否有 tolist 方法（QMT 返回的可能是 numpy 数组）
if hasattr(close_data, 'tolist'):
    close_list = close_data.tolist()
else:
    close_list = close_data

# 从一个字典筛选出满足条件的键值对
all_growth = {"A": 5.2, "B": -3.1, "C": 8.7, "D": -1.0}

positive = {k: v for k, v in all_growth.items() if v > 0}
# {"A": 5.2, "C": 8.7}

count_dict = {}       # 行业 -> 已选数量
final_stocks = []      # 最终选中的股票

for stock in stock_list:
    industry = Context.code2industry_dict[stock]   # 查字典获取行业

    if industry not in count_dict:
        count_dict[industry] = 1       # 第一次出现
        final_stocks.append(stock)
    elif count_dict[industry] < 3:     # 每行业最多3只
        count_dict[industry] += 1
        final_stocks.append(stock)
    # 超过3只的跳过

# 风格判断规则矩阵
if small_ret >= -4:
    print("小盘偏强 → 执行小市值选股")
    target_stocks = get_target_stocks(ContextInfo, current_date)

elif small_ret < -4 and big_ret > 0:
    print("大盘偏强 → 红利ETF+银行ETF")
    # 配置防御性资产

elif small_ret < -4 and big_ret < 0:
    print("全市场弱 → 黄金ETF+国债ETF")
    # 配置避险资产

else:
    print("边界情况，不调仓")
    return

# 遍历列表
stocks = ["600519.SH", "000858.SZ", "300750.SZ"]
for stock in stocks:
    print(f"处理 {stock}")

# 遍历字典
holdings = {"600519.SH": 5, "000858.SZ": 10}
for stock, hand in holdings.items():
    print(f"{stock} 持有 {hand} 手")

# 带索引的遍历
for i, stock in enumerate(stocks):
    print(f"第{i+1}只: {stock}")

# range 生成数字序列
for i in range(5):       # 0, 1, 2, 3, 4
    print(i)
for i in range(1, 6):    # 1, 2, 3, 4, 5
    print(i)

filtered = []
for stock in stock_list:
    try:
        # 停牌过滤
        if ContextInfo.is_suspended_stock(stock):
            continue    # 跳过这只，继续下一只

        # ST过滤
        stock_name = ContextInfo.get_stock_name(stock)
        if "ST" in stock_name or "退" in stock_name:
            continue    # 跳过

        # 通过所有过滤条件
        filtered.append(stock)

    except Exception as e:
        print(f"过滤{stock}时出错：{str(e)}")
        continue        # 出错也跳过，不中断整个循环

# lambda 参数: 返回值
# 等价于：def func(x): return x[1]

data = [("A", 50), ("B", 20), ("C", 80)]

# 按第二个元素升序排列
data.sort(key=lambda x: x[1])
# [("B", 20), ("A", 50), ("C", 80)]

# 按第二个元素降序排列
data.sort(key=lambda x: x[1], reverse=True)
# [("C", 80), ("A", 50), ("B", 20)]

# 定义函数：def 函数名(参数):
def calculate_high_limit(pre_close, code_part):
    """计算涨停价"""
    if pre_close <= 0:
        return 0

    if code_part.startswith("68") or code_part.startswith("30"):
        return round(pre_close * 1.20, 2)   # 20%涨跌幅
    else:
        return round(pre_close * 1.10, 2)   # 10%涨跌幅

# 调用函数
price1 = calculate_high_limit(14.0, "300001")   # 16.8
price2 = calculate_high_limit(1700.0, "600519") # 1870.0

# 返回单个值
def get_pre_close(stock):
    # ... 获取前收盘价的逻辑
    return 15.68

# 返回多个值（实际上是返回元组）
def get_style_returns(ContextInfo, current_date):
    big_ret = 3.5
    small_ret = -5.2
    return (big_ret, small_ret)

# 调用时用多个变量接收
big_ret, small_ret = get_style_returns(ContextInfo, current_date)

# 不返回值，只执行操作
def check_limit_up(ContextInfo):
    # 检查涨停状态，只打印信息
    print("涨停炸板检查...")
    # 没有 return，默认返回 None

# 必需参数：调用时必须传入
def filter_st(stock_list, ContextInfo):
    ...

# 带默认值的参数：调用时可以不传，使用默认值
def calc_atr(stock, ContextInfo, period=20):
    ...

# 调用
calc_atr("600519.SH", ContextInfo)           # period 使用默认值 20
calc_atr("600519.SH", ContextInfo, period=10) # period 使用传入值 10

def get_style_returns(ContextInfo, current_date):
    """风格判断：计算大盘和小盘的N日涨幅"""

    # 内部函数：只在 get_style_returns 内部使用
    def calc_index_return(index_code):
        try:
            hist_data = ContextInfo.get_market_data_ex(...)
            # ... 计算逻辑
            return 涨幅
        except:
            return 0.0

    # 调用内部函数
    big_ret = calc_index_return("000300.SH")    # 沪深300
    small_ret = calc_index_return("399101.SZ")  # 中小板综
    return (big_ret, small_ret)

# 基本结构
try:
    # 可能出错的代码
    result = 10 / 0
except Exception as e:
    # 出错时执行
    print(f"出错了：{str(e)}")
# 输出：出错了：division by zero

def get_last_price(ContextInfo, stock):
    """获取最新价"""
    try:
        tick_data = ContextInfo.get_full_tick([stock])
        if tick_data and tick_data[stock] and tick_data[stock]["lastPrice"] > 0:
            return tick_data[stock]["lastPrice"]
    except Exception as e:
        print(f"获取{stock}最新价时出错：{str(e)}")

    return 0   # 出错时返回默认值

try:
    detail = ContextInfo.get_instrument_detail(stock)
    # ... 处理逻辑
except:
    pass    # 出错时什么都不做，直接跳过

import datetime

# 字符串 → 日期对象
date_str = "20260718"
date_obj = datetime.datetime.strptime(date_str, "%Y%m%d")
# datetime.datetime(2026, 7, 18, 0, 0)

# 日期对象 → 字符串
formatted = date_obj.strftime("%Y%m%d")
# "20260718"

# 日期计算
from datetime import datetime, timedelta
today = datetime(2026, 7, 18)
yesterday = today - timedelta(days=1)     # 2026-07-17
one_year_ago = today - timedelta(days=365)  # 2025-07-19

# 计算两个日期之间的天数差
open_date = datetime(2025, 3, 15)
days_since_listing = (today - open_date).days   # 490

# 获取星期几（0=周一，6=周日）
weekday = today.weekday()    # 5（周六）
is_monday = (weekday == 0)   # False

# 获取周数
week_num = today.isocalendar()[1]   # 29（第29周）

# 导入整个模块，使用时需要写模块名
import datetime
datetime.datetime.strptime(...)

# 导入模块中的特定内容，使用时不需要写模块名
from datetime import datetime, timedelta
datetime.strptime(...)      # 不需要写 datetime.datetime
timedelta(days=1)           # 直接使用

# 导入并取别名（numpy 常用）
import numpy as np
np.mean([1, 2, 3])   # 2.0

# 检查对象是否有某个属性
if hasattr(pos, 'm_strInstrumentID') and hasattr(pos, 'm_strExchangeID'):
    stock = f"{pos.m_strInstrumentID}.{pos.m_strExchangeID}"
elif hasattr(pos, 'code'):
    stock = pos.code

# 检查 numpy 数组是否有 tolist 方法
if hasattr(close_data, 'tolist'):
    close_list = close_data.tolist()
else:
    close_list = close_data

# === 这些都是全局变量，在策略启动时执行一次 ===

# Context 是 QMT 框架提供的全局对象
Context.accountID = "你的资金账号"
Context.stock_num = 10           # 持仓股票数量
Context.style_lookback_days = 20 # 风格判断回看天数
Context.small_ret_threshold = -4 # 小盘涨幅阈值
Context.atr_period = 20          # ATR 周期
Context.atr_multiple = 2         # ATR 倍数
Context.enable_atr_stop_loss = True  # 是否启用ATR止损
Context.max_industry_count = 3   # 单行业最大持仓数

# 运行时状态变量
Context.last_check_date = ""     # 上次检查日期
Context.last_week = 0            # 上次周数
Context.hold_list = []           # 当前持仓列表
Context.high_limit_list = []     # 昨日涨停列表

# 防御性资产配置（字典 + 列表）
Context.defensive_assets = {
    "大盘偏强": [
        {"code": "510050.SH", "weight": 0.4},  # 上证50ETF
        {"code": "510880.SH", "weight": 0.4},  # 红利ETF
        {"code": "600900.SH", "weight": 0.2},  # 长江电力
    ],
    "全市场弱": [
        {"code": "518880.SH", "weight": 0.5},  # 黄金ETF
        {"code": "511010.SH", "weight": 0.5},  # 国债ETF
    ]
}

# 行业分类字典（股票代码 → 行业名称），数据量很大
Context.code2industry_dict = {
    '000001.SZ': 'SW1银行',
    '000002.SZ': 'SW1房地产',
    # ... 几千条 ...
}

def handlebar(ContextInfo):
    """核心执行函数，每根K线执行一次"""

    # 获取当前K线位置
    d = ContextInfo.barpos
    if d < Context.style_lookback_days + 5:  # 数据不足，跳过
        return

    # 获取当前日期 → 字符串操作
    current_timetag = ContextInfo.get_bar_timetag(d)
    current_date_str = timetag_to_datetime(current_timetag, "%Y%m%d")

    # 判断是否新交易日 → 字符串比较
    is_new_day = current_date_str != Context.last_check_date
    if not is_new_day:
        return

    Context.last_check_date = current_date_str

    # 判断是否新的一周 → datetime 日期计算
    if prev_date_str:
        current_date_obj = datetime.datetime.strptime(current_date_str, "%Y%m%d")
        prev_date_obj = datetime.datetime.strptime(prev_date_str, "%Y%m%d")

        is_new_week = (current_date_obj.weekday() == 0) or \
                      (current_date_obj - prev_date_obj).days > 3

        current_week = current_date_obj.isocalendar()[1]
        if current_week != Context.last_week:
            is_new_week = True
            Context.last_week = current_week

    # 每日执行：风控
    check_limit_up(ContextInfo)       # 涨停炸板检查
    check_atr_stop_loss(ContextInfo)  # ATR止损检查

    # 每周执行：调仓
    if is_new_week:
        weekly_adjustment(ContextInfo)

def filter_st(stock_list, ContextInfo):
    """ST过滤"""
    filtered = []                          # 空列表
    for stock in stock_list:               # for 循环遍历
        try:                               # 异常处理
            stock_name = ContextInfo.get_stock_name(stock)
            if stock_name:                 # 非空判断
                if "ST" not in stock_name and "*ST" not in stock_name \
                   and "退" not in stock_name:   # 字符串 in 判断
                    filtered.append(stock)       # 列表添加
            else:
                filtered.append(stock)           # 名称为空，保留
        except:
            filtered.append(stock)               # 出错保留
    return filtered

def filter_new(stock_list, ContextInfo, current_date):
    """次新股过滤（上市不足365天）"""
    filtered = []
    current_dt = datetime.datetime.strptime(current_date, "%Y%m%d")  # 日期解析

    for stock in stock_list:
        try:
            detail = ContextInfo.get_instrument_detail(stock)   # QMT API
            if detail and "OpenDate" in detail:                 # 字典键检查
                open_date_str = str(detail["OpenDate"])         # 转字符串
                if open_date_str.isdigit() and len(open_date_str) == 8:
                    open_date = datetime.datetime.strptime(open_date_str, "%Y%m%d")
                    days_diff = (current_dt - open_date).days    # 日期差
                    if days_diff >= 365:                         # 比较
                        filtered.append(stock)
                else:
                    filtered.append(stock)
            else:
                filtered.append(stock)
        except:
            filtered.append(stock)
    return filtered

# 卖出不在目标中的持仓
for stock, hand in holdings.items():          # 遍历字典
    if hand <= 0:
        continue                              # 持仓为0跳过

    if stock not in target_assets:            # 字典键检查
        # 涨停暂不卖出
        if stock in Context.high_limit_list:   # 列表包含检查
            print(f"{stock} 昨日涨停，暂不卖出")
            continue

        # 卖出
        price = get_last_price(ContextInfo, stock)  # 调用函数
        if price > 0:
            volume = -hand * 100                     # 算术运算
            order_shares(stock, volume, "latest", price,
                        ContextInfo, Context.accountID)  # QMT 下单API

# 买入目标资产
for stock, target_weight in target_assets.items():
    # 计算目标市值
    target_value = total_asset * target_weight

    price = get_last_price(ContextInfo, stock)
    if price <= 0:
        continue

    # 非ETF需要额外检查
    if not is_etf(stock):
        if ContextInfo.is_suspended_stock(stock):
            continue

        pre_close = get_pre_close(ContextInfo, stock, current_date)
        if pre_close > 0 and is_limit_up(stock, price, pre_close):
            print(f"{stock} 涨停，跳过买入")
            continue

    # 计算买入股数（100的整数倍）
    volume = int(target_value / price / 100) * 100   # 整除

    if volume >= 100:
        order_shares(stock, volume, "latest", price,
                    ContextInfo, Context.accountID)

def calc_atr(stock, ContextInfo):
    """计算ATR（Average True Range）"""
    period = Context.atr_period

    # 获取历史数据（QMT API）
    hist_data = ContextInfo.get_market_data_ex(
        ["high", "low", "close"], [stock],
        period="1d", count=period + 1
    )

    # 提取数据（处理 numpy 数组）
    high_data = hist_data[stock]["high"]
    if hasattr(high_data, 'tolist'):       # hasattr 属性检查
        highs = high_data.tolist()
    else:
        highs = high_data

    # 计算 TR（True Range）
    tr_values = []
    for i in range(1, len(highs)):          # for + range 循环
        high = float(highs[i])
        low = float(lows[i])
        prev_close = float(closes[i - 1])

        tr1 = high - low                    # 当日振幅
        tr2 = abs(high - prev_close)        # 当日最高-昨收
        tr3 = abs(low - prev_close)         # 当日最低-昨收

        tr = max(tr1, tr2, tr3)             # 取最大值
        tr_values.append(tr)

    # 计算ATR（最近period个TR的平均值）
    atr = np.mean(tr_values[-period:])      # numpy求均值 + 列表切片
    return atr