昨天我们介绍了怎么样利用AI工具workbuddy调用skill实现我们的策略

量化研究--workbuddy量化skill使用，

今天我们调用多个模块实现，一个全球大类的的动量轮动策略，当然也有回测的我无业版本比较方便

第一步是在ai立马安装skill工具，这个是我开发的方便做投资研究

安装自己感兴趣的模块就可以比如我安装的，投资研究的矩阵

我们只需要在对话框加/就可以选择自己的skill，多个skill用英文的逗号隔开就可以

比如我同时调用2个模型，一个是数据模块，一个是动量回测系统

给ai提示内容同时调用2个skill分析策略


/小果量化数据助手 ，/小果动量策略回测系统助手 ，
利用小果数据助手，读取标的的行情数据开始时间20250101，
结束时间20260902，利用动量回测系统，
把读取到的pandas数据添加进入回测系统，
实现全球大类资产etf的动量回测的代码，给我输出完整的代码，
必须要分析过程，每日计算每个标的在过去25个交易日的动31期间，
以000300.SH为基准，
对1515.SZ、5100.SH、880.SH进行动量排名1。交易佣金按1万计算，初始资金百分比，每次交易金额/数量为1。当单笔持仓收益率≥3%时触发止盈卖出，卖出金额为:
策略思路:本策略基于 「动量轮动+趋势跟踪」逻辑，在2025-01-01至2026-12-
量分数，并筛选出分数在0至5之间的标的，
只持有买入排名前1的标的。交易模式为
¥100.000。策略旨在通过动量效应捕捉强势标的，
实现资产的轮动增值。

ai输出的内容本地运行看看，mcp个人不能发布不然必须要本地运行

运行的结构没有问题，目前skill是第一个版本，后面再优化一下更强大

代码版本的是给ai研究的，我一般使用网页版本研究方便，操作简单。结果是一样的


# -*- coding: utf-8 -*-
"""
=============================================================================
 全球大类资产 ETF「动量轮动 + 趋势跟踪」回测
=============================================================================
数据源   : 小果量化数据助手   xg_quant_backtrader_data  (读取行情 pandas)
回测引擎 : 小果动量策略回测系统 xg_mom_backtrader        (add_stock_data_from_dataframe 注入)
作者     : 小果量化 (微信: xg_quant)
一、策略说明(逐日分析过程)
--------------------------------------------------------------------------------
标的池   : 155.SZ(创ETF-中国成长) / 500.SH(ETF股科技)
           / 518.SH(ETF-避险商品)          -> 全球大类资产分散轮动
基准     : 000300.SH (沪深300)  [可选: 指数数据读取失败/为空时自动跳过,
           不影响回测引擎运行, 仅3只ETF行情齐全即可正常回测]
回测区间 : 20250101 ~ 20260902
初始资金 : ¥100,000     交易佣金 : 万1 (0.0001)
每个交易日系统依次执行以下分析步骤:
① 动量打分  : 对每只标的取“过去25个交易日”(mom_daily=25)的收盘价做
              【加权线性回归(权重1->2递增,近期权重大)】, 斜率年化后 × R²
              得到动量分数 score = 年化收益 × R²  (引擎 calculate_momentum_score)
② 区间筛选  : 仅保留动量分数落在 [0, 5] (min_mom/max_mom) 区间的标的,
              过滤过热(>5追高)与弱势(<0下行)标的;
              若当日全部落榜则回退取排名第1, 保证始终跟踪最强趋势
③ 动量排名  : 按分数降序排名, 只持有/买入排名前 buy_rank=1 名的标的
④ 先卖后买  : 持仓跌出前1 -> 全仓动量卖出; 新晋第1 -> 动用总资产100%
              (mom_type='百分比', mom_value=1.0) 满仓买入, 轮动切换
⑤ 止盈卖出  : 持仓当日收益(涨幅) ≥ 3% (sell_zdf=0.03) -> 按 sell_amount
              金额止盈卖出(卖出金额取≈初始资金, 触发即基本清仓锁利)
二、运行环境
--------------------------------------------------------------------------------
pip install requests pandas numpy pyarrow openpyxl
并保证可 import:
  from xg_quant_backtrader_data.xg_quant_backtrader_data import xg_quant_backtrader_data
  from xg_mom_backtrader.xg_mom_backtrader import xg_mom_backtrader
================================================================================
"""
import sys
import pandas as pd
import numpy as np
# =============================================================================
# 1. 参数配置区(★按需修改)
# =============================================================================
# ---------- 小果量化数据服务器配置(联系小果开通) ----------
DATA_SERVER = dict(
    url="124.220.32.224",     # ★ 你的服务器地址
    port=8888,
    user="小果",     # ★ 你的用户名
    password="小果", # ★ 你的密码
    auth_code="小果",  # ★ 你的授权码
)
# ---------- 回测参数 ----------
START_DATE   = "20250101"                     # 行情/回测开始
END_DATE     = "20260902"                     # 行情/回测结束
STOCK_LIST   = ["915.SZ", "100.SH", "880.SH"]  # 全球大类资产ETF池
INDEX_STOCK  = "000300.SH"                    # 基准指数(沪深300)
CASH         = 100000.0                       # 初始资金 ¥100,000
COMM         = 0.0001                         # 交易佣金 万1
# 动量/排名/交易
MOM_TYPE     = "百分比"                       # 资金分配方式: 百分比/金额
MOM_VALUE    = 1.0                            # 每次交易动用总资产 100%(即"每次交易金额/数量为1")
MOM_DAILY    = 25                             # 过去25个交易日动量(加权线性回归窗口)
MIN_MOM      = 0.0                            # 动量分数下限(0~5区间筛选)
MAX_MOM      = 5.0                            # 动量分数上限
BUY_RANK     = 1                              # 只持有/买入动量排名第1
# 止盈
SELL_ZDF     = 0.03                           # 当日收益≥3% 触发止盈
SELL_AMOUNT  = CASH                           # 止盈卖出金额(默认≈初始资金,触发即清仓;可按需改小做部分止盈)
MAX_WORKERS  = 8                              # 多线程加载线程数
PLOT         = True                           # 是否画净值对比图(需matplotlib)
# =============================================================================
# 2. 小果数据助手: 读取标的 + 基准行情 (20250101 ~ 20260902)
# =============================================================================
from xg_quant_backtrader_data.xg_quant_backtrader_data import xg_quant_backtrader_data
def fetch_hist(client, code):
    """读取单只标的日线, 返回标准OHLCV DataFrame"""
    result = client.get_stock_hist_data(
        stock=code, start_date=START_DATE, end_date=END_DATE
    )
    df = client._to_dataframe(result)          # 接口统一转 pandas
    if df is None or df.empty:
        return pd.DataFrame()
    # 统一列名 / 数值化, 确保引擎必需字段 date,open,high,low,close,volume 齐全
    rename_map = {}
    for col in df.columns:
        low = str(col).lower()
        if "date" in low:
            rename_map[col] = "date"
        elif low in ("open", "high", "low", "close", "volume", "amount"):
            rename_map[col] = low
    df = df.rename(columns=rename_map)
    for col in ["date", "open", "high", "low", "close", "volume"]:
        if col not in df.columns:
            df[col] = np.nan
    df["date"] = pd.to_datetime(df["date"])
    df = df.sort_values("date").drop_duplicates("date")
    return df
def main():
    print("=" * 60)
    print("📡 ① 初始化小果量化数据助手并读取全球资产行情")
    print("=" * 60)
    client = xg_quant_backtrader_data(**DATA_SERVER)
    # ---- ① 读取 3 只 ETF (必须齐全, 否则终止) ----
    data_map = {}
    for code in STOCK_LIST:
        df = fetch_hist(client, code)
        data_map[code] = df
        if df.empty:
            print(f"  ❌ {code} 行情读取为空, 请检查服务器配置/授权码!")
        else:
            print(f"  ✅ {code}: {len(df)} 行, "
                  f"{df['date'].min().date()} ~ {df['date'].max().date()}, "
                  f"close {df['close'].iloc[0]:.4f} -> {df['close'].iloc[-1]:.4f}")
    etf_empty = [c for c, df in data_map.items() if df.empty]
    if etf_empty:
        sys.exit(f"ETF行情缺失 {etf_empty}, 终止回测")
    # ---- ② 读取基准指数 000300.SH (可选, 失败不影响回测) ----
    idx_df = fetch_hist(client, INDEX_STOCK)
    if idx_df.empty:
        print(f"  ⚠️ {INDEX_STOCK} 指数数据未读取到, 本次回测不使用指数对比, 不影响运行")
    else:
        print(f"  ✅ {INDEX_STOCK}: {len(idx_df)} 行, "
              f"{idx_df['date'].min().date()} ~ {idx_df['date'].max().date()}, "
              f"close {idx_df['close'].iloc[0]:.4f} -> {idx_df['close'].iloc[-1]:.4f}")
    # =========================================================================
    # 3. 小果动量回测系统: 构造引擎(子类: 使 000300.SH 基准指数也能走内存数据)
    # =========================================================================
    print("\n" + "=" * 60)
    print("📊 ② 初始化小果动量策略回测系统")
    print("=" * 60)
    from xg_mom_backtrader.xg_mom_backtrader import xg_mom_backtrader
    class GlobalMomBT(xg_mom_backtrader):
        """基于引擎的轻量子类:
           1) 引擎原版 __init__ 未定义 self.verbose, 这里补上(否则注入接口报错)
           2) 重写 get_index_data: 优先取注入的指数DataFrame; 未注入则回退本地parquet,
              都没有时返回空 DataFrame —— 引擎对此自动跳过, 不影响回测运行
        """
        def __init__(self, *args, **kwargs):
            super().__init__(*args, **kwargs)
            self.verbose = False
        def get_index_data(self):
            try:
                if self.has_custom_data(self.index_stock):
                    df = self._custom_data_cache[self.index_stock].copy()
                    df["date"] = pd.to_datetime(df["date"])
                    s, e = pd.to_datetime(self.start_date), pd.to_datetime(self.end_date)
                    df = df[(df["date"] >= s) & (df["date"] <= e)].sort_values("date")
                    df = df[(df["close"] > 0) & (df["open"] > 0)].reset_index(drop=True)
                    print(f"指数数据(内存注入)加载成功: {len(df)} 行")
                    return df
                # 未注入指数 -> 尝试本地 parquet, 没有则返回空(引擎自动跳过, 不影响回测)
                return super().get_index_data()
            except Exception as ex:
                print(f"加载指数数据出错(已忽略, 不影响回测): {ex}")
                return pd.DataFrame()
    bt = GlobalMomBT(
        start_date=START_DATE,
        end_date=END_DATE,
        stock_list=STOCK_LIST,          # 注意: 引擎 stock_list 参数用 list
        index_stock=INDEX_STOCK,
        cash=CASH,
        comm=COMM,
        mom_type=MOM_TYPE,
        mom_value=MOM_VALUE,            # 每次动用总资产100%
        mom_daily=MOM_DAILY,            # 过去25个交易日动量
        min_mom=MIN_MOM,                # 分数区间下限 0
        max_mom=MAX_MOM,                # 分数区间上限 5
        buy_rank=BUY_RANK,              # 只持有第1名
        sell_zdf=SELL_ZDF,              # 收益≥3%触发止盈
        sell_amount=SELL_AMOUNT,        # 止盈卖出金额
        max_workers=MAX_WORKERS,
    )
    # =========================================================================
    # 4. 把小果数据助手读到的 pandas 数据逐只注入回测系统
    # =========================================================================
    print("\n" + "=" * 60)
    print("💉 ③ 注入 pandas 行情数据到动量回测系统")
    print("=" * 60)
    for code in STOCK_LIST:
        ok = bt.add_stock_data_from_dataframe(code, data_map[code])
        print(f"  {'✅' if ok else '❌'} add_stock_data_from_dataframe: {code}")
    if not idx_df.empty:
        bt.add_stock_data_from_dataframe(INDEX_STOCK, idx_df)   # 指数仅用于对比展示,可选
    print("已注入标的:", bt.get_custom_data_keys())
    # =========================================================================
    # 5. 执行动量回测
    # =========================================================================
    bt.run_backtest()
    print("\n" + "=" * 60)
    print("📈 回测完成, 详细报告如下")
    print("=" * 60)
    print(bt.generate_report())
    # ---------- 核心绩效汇总 ----------
    m = bt.get_performance_metrics()
    print("\n【核心绩效汇总】")
    print(f"  区间       : {m['start_date']} ~ {m['end_date']}")
    print(f"  初始资金   : ¥{m['total_cash']:,.2f}")
    print(f"  期末总资产 : ¥{m['final_value']:,.2f}")
    print(f"  总收益率   : {m['total_return'] * 100:.2f}%")
    print(f"  年化收益率 : {m['annual_return'] * 100:.2f}%")
    print(f"  最大回撤   : {m['max_drawdown'] * 100:.2f}% ({m.get('max_drawdown_date')})")
    print(f"  夏普比率   : {m['sharpe_ratio']:.3f}")
    print(f"  胜率       : {m['win_rate'] * 100:.2f}%")
    print(f"  总手续费   : ¥{m['total_commission']:,.2f}")
    print(f"  超额收益   : {m['excess_return_pct']:.2f}%  (vs 三ETF等权持有不动)")
    # =========================================================================
    # 6. 结果落盘 (Excel + JSON) 与净值曲线图 (任一步失败均不影响其余输出)
    # =========================================================================
    try:
        bt.save_backtrader_data(user="全球动量ETF轮动")   # 保存到 策略模型/动量策略/全球动量ETF轮动/
    except Exception as e:
        print(f"  ⚠️ 分目录保存(Excel/JSON)跳过: {e}")
    try:
        bt.save_to_json("momentum_global_etf_result.json", include_raw_data=False)
    except Exception as e:
        print(f"  ⚠️ 汇总JSON保存跳过: {e}")
    curve = bt.get_equity_curve_data()
    dates, net = curve["dates"], curve["net_value"]
    bench = curve.get("benchmark_net_value", [])
    # ---- 以沪深300为对比基准, 单独归一化一条曲线 (指数未读到则自动跳过) ----
    idx_nav = []
    if not idx_df.empty:
        tmp = idx_df[idx_df["date"].isin(pd.to_datetime(dates))].reset_index(drop=True)
        if len(tmp):
            idx_nav = (tmp["close"] / tmp["close"].iloc[0]).round(4).tolist()
    if PLOT:
        try:
            import matplotlib
            matplotlib.use("Agg")
            import matplotlib.pyplot as plt
            plt.rcParams["font.sans-serif"] = ["SimHei", "Microsoft YaHei", "Arial Unicode MS"]
            plt.rcParams["axes.unicode_minus"] = False
            fig, (ax1, ax2) = plt.subplots(
                2, 1, figsize=(12, 7),
                gridspec_kw={"height_ratios": [3, 1]}, sharex=True)
            ax1.plot(dates, net, label="动量轮动策略", lw=2, color="#c0392b")
            if bench:
                ax1.plot(dates, bench, label="三ETF等权持有(引擎基准)", ls="--", lw=1.2)
            if idx_nav and len(idx_nav) == len(dates):
                ax1.plot(dates, idx_nav, label="沪深300(000300.SH)", ls="-.", lw=1.2)
            ax1.set_title("全球大类资产ETF 动量轮动回测净值")
            ax1.legend(loc="upper left"); ax1.grid(alpha=0.3); ax1.set_ylabel("净值")
            dd = curve.get("drawdown", [])
            ax2.fill_between(dates, 0, dd, color="#c0392b", alpha=0.3)
            ax2.plot(dates, dd, color="#c0392b", lw=1)
            ax2.set_ylabel("回撤(%)"); ax2.grid(alpha=0.3)
            for ax in (ax1, ax2):
                for lab in ax.get_xticklabels():
                    lab.set_rotation(45); lab.set_fontsize(8)
            plt.tight_layout()
            plt.savefig("momentum_global_etf_curve.png", dpi=130)
            print("\n✅ 净值曲线已保存: momentum_global_etf_curve.png")
        except Exception as e:
            print(f"⚠️ 绘图跳过(未安装matplotlib或字体问题): {e}")
    print("\n✅ 全流程执行完毕")
if __name__ == "__main__":
    main()

AI同时调用多个量化Skill，动量策略一句话跑通

投资顾问

人工智能

算法

利用AI工具调用小果量化skill实现全球大类资产ETF动量轮动策略，通过25日加权动量打分与趋势跟踪筛选，结合排名轮动和3%止盈机制，在2025-2026年回测区间内自动完成资产配置切换，实现收益优化。

数据库

服务器

JSON

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

AI同时调用多个量化Skill，动量策略一句话跑通

AI同时调用多个量化Skill，动量策略一句话跑通

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐