蜗牛之路

QMT自带函数说明，可以查看一些函数的使用方法。

而且有示例代码，可以拿来直接测试。

一、获取股票历史信息

import pandas as pd
import numpy as np
import talib

def init(ContextInfo):
    ContextInfo.set_universe(['600300.SH','000004.SZ'])
def handlebar(ContextInfo):
    #获取股票池中所有股票的最近两日的收盘价
    hisdict = ContextInfo.get_history_data(10,'1d','close')
    print(hisdict)
    for k, v in hisdict.items():
        if len(v)>1:
            print( k,':',v[1]-v[0])#今日涨幅

比如上面的代码，就是取到了股票近10天的收盘价。

二、获取股票的1分钟数据

# encoding: utf-8

import pandas as pd
import numpy as np
import talib

def init(ContextInfo):
    pass
    
    
def handlebar(ContextInfo):
    #获取股票池中股票的最近1日的4价
    hisdict =ContextInfo.get_market_data_ex(
    ['open', 'high', 'low', 'close'], ['688062.SH'], period='1m'
    , start_time='', end_time='', count=-1
    , dividend_type='follow', fill_data=True
    , subscribe = True
    )
    print(hisdict)

经核对没有任何问题。

如果要获得成交果，只需要在列表中加入即可。即： ['open', 'high', 'low', 'close','volume']

经查看函数使用说明，count：数据最大个数，-1视为不做个数限制，不过将这个数改成480，也没有用，返回的还是240，也就是说它只能获取1天的1m数据。

三、计算指标

# encoding: gbk
import pandas as pd
import datetime

def init(ContextInfo):
    """
    初始化函数，设定每天 15:05 触发策略。
    """
    ContextInfo.run_time(["15:05"])  # ✅ 正确方式：设定每天 15:05 运行一次


def handlebar(ContextInfo):
    """
    在 15:05 自动执行的主逻辑。
    """
    analyze_day_volume(ContextInfo)


def analyze_day_volume(ContextInfo):
    """
    分析尾盘成交量比值并保存为 CSV。
    """
    stock_list = ['688062.SH', '600519.SH', '000001.SZ']  # ✅ 替换为你的股票池
    results = {}

    # 获取当日所有股票的 1 分钟数据
    hisdict = ContextInfo.get_market_data_ex(
        ['open', 'high', 'low', 'close', 'volume'],
        stock_list,
        period='1m',
        start_time='', end_time='', count=-1,
        dividend_type='follow', fill_data=True,
        subscribe=False
    )

    for symbol, df in hisdict.items():
        try:
            ratio = calculate_tail_volume_ratio(df)
            results[symbol] = ratio
        except Exception as e:
            print(f"{symbol} 分析失败: {e}")
            results[symbol] = None

    # 保存结果为 CSV 文件
    date_str = datetime.datetime.now().strftime("%Y%m%d")
    df_result = pd.DataFrame.from_dict(results, orient='index', columns=['prev_tail_vol_ratio'])
    df_result.index.name = 'symbol'
    filename = f"prev_tail_vol_ratio_{date_str}.csv"
    df_result.to_csv(filename, encoding='utf-8-sig')
    print(f"[{date_str}] 分析完成，已保存至：{filename}")


def calculate_tail_volume_ratio(df: pd.DataFrame) -> float:
    """
    计算尾盘成交量比值: (14:30-15:00 成交量) / (10:00-10:30 成交量)
    """
    df = df.copy()
    df.index = pd.to_datetime(df.index, format="%Y%m%d%H%M%S")
    morning = df.between_time("10:00", "10:29")
    afternoon = df.between_time("14:30", "14:59")
    morning_volume = morning["volume"].sum()
    afternoon_volume = afternoon["volume"].sum()
    if morning_volume == 0:
        return float("inf")
    return afternoon_volume / morning_volume