声明：本内容为个人业余研究，所有标的代码仅做为示例，回测收益不代表未来，不做为投资建议。

目前是根据球友的建议，先选取了几只etf，AI大模型炒股项目的雏形可以跑起来了。用qwen3和deepseek做了交易指令，qwen3是有交易记录， Deepseek始终保持持仓不变，一直没有出手,后面拉长了回测时间，增加etf数量之后再看是否会有交易信号。股票后面也可以支持的，只需要把\configs\default_config.json中的all_zh_symbols换成股票的代码，提前下载股票的行情数据就行。本次主要修改点在于下载etf的行情数据，让大模型读取ETF行情数据，修改回测基准指标为沪深 300。 通过整合akshare金融数据接口库，获取中国市场各类ETF（交易所交易基金）的实时行情数据，实现了对多市场ETF数据的自动化采集、处理和存储功能。考虑了数据持久化需求，能够智能识别已存在的数据文件并实现增量更新，避免重复获取历史数据。

环境配置与初始化

import osfrom dotenv import load_dotenvload_dotenv()import jsonimport akshare as akfrom datetime import datetime, timedelta

这部分代码完成了项目的初始设置工作。首先导入必要的标准库：os用于文件系统操作，dotenv用于加载环境变量，json处理JSON格式数据，datetime处理时间相关计算。特别值得注意的是load_dotenv()函数调用，它会从项目根目录的.env文件中加载环境变量，这种方式避免了将API密钥等敏感信息硬编码在源代码中。

akshare库是本项目的核心依赖，它是一个免费开源的金融数据接口库，提供了丰富的中国股市数据接口。相比其他商业数据源，akshare具有无需付费、接口稳定且更新及时的优势，非常适合个人投资者使用。

股票代码列表定义

all_zh_symbols = [    # 海外市场    "159561", # 嘉实德国DAX ETF    "513880", # 日经225ETF    "513500", # 标普500ETF    # 香港市场    "513130",  # 恒生科技ETF    "513830",  # 港股红利ETF    # 资源类    "160416",  # 华安标普全球石油指数LOF    "518880",  # 黄金ETF]

这个列表定义了项目监控的所有ETF产品代码。每个注释都标明了对应ETF的具体名称和投资方向。从列表可以看出，项目选择了具有代表性的跨境ETF产品，包括发达国家市场指数ETF、新兴市场ETF以及大宗商品类ETF，形成了较为完整的资产配置组合。

选择这些特定ETF而非直接购买个股，主要是因为ETF具有分散风险、费率低廉、透明度高等优点，特别适合普通投资者。同时，跨境ETF还能够帮助投资者实现资产国际化配置，降低单一市场风险。

股票代码格式转换函数

def add_sh_sz(symbol):    if symbol.startswith(("6","5")):        symbol = f"sh{symbol}"    elif symbol.startswith(("0","1","3")):         symbol = f"sz{symbol}"    return symbol

该函数负责将原始股票代码转换为akshare API要求的格式。在中国证券市场中，上海证券交易所的股票代码以6开头，深圳证券交易所的股票代码以0、1或3开头。函数根据这一规则，在原始代码前添加相应的交易所标识符（sh表示上海，sz表示深圳）。

例如，输入"510300"会返回"sh510300"，而输入"159941"则会返回"sz159941"。这种格式转换是调用akshare接口的必要步骤，因为不同的交易所需要不同的标识来区分。

数据刷新日期检测函数

def get_last_refreshed_date(SYMBOL: str):    try:        filename = f'./daily_prices_{SYMBOL}.json'        if os.path.exists(filename):            with open(filename, 'r', encoding='utf-8') as f:                data = json.load(f)                last_refreshed = data.get("Meta Data", {}).get("3. Last Refreshed")                if last_refreshed:                    return datetime.strptime(last_refreshed, "%Y-%m-%d")        return None    except Exception as e:        print(f"读取现有文件失败: {e}")        return None

此函数的主要作用是检查指定ETF是否已有本地存储的历史数据文件，并从中提取最后更新日期。函数首先构建预期的文件名格式，然后检查该文件是否存在。如果存在，则打开文件并解析其中的JSON数据，定位到"Meta Data"部分下的"Last Refreshed"字段。

成功获取该日期字符串后，函数将其转换为datetime对象以便进行日期计算。如果过程中出现任何异常（如文件损坏、格式错误等），函数会捕获异常并返回None，同时打印错误信息供调试使用。

这种机制使得程序能够智能判断是否需要下载新数据，对于已经拥有历史数据的ETF，可以从上次结束的地方继续获取最新数据，大大提高了效率并减少了不必要的网络请求。

主数据获取函数

def get_daily_price(SYMBOL: str):    try:        last_refreshed_date = get_last_refreshed_date(SYMBOL)                if last_refreshed_date:            start_date = (last_refreshed_date + timedelta(days=1)).strftime("%Y%m%d")            print(f"发现历史数据，最新刷新时间: {last_refreshed_date.strftime('%Y-%m-%d')}")            print(f"从 {start_date} 开始获取新数据")        else:            start_date = "20251101"            print(f"未发现历史数据，从 {start_date} 开始获取数据")                end_date = datetime.now().strftime("%Y%m%d")        print(f"获取数据时间范围: {start_date} 到 {end_date}")                df = ak.fund_etf_hist_sina(symbol=add_sh_sz(SYMBOL))        print(f"获取到的数据形状: {df.shape}")        print(f"数据列名: {df.columns.tolist()}")        print(f"索引类型: {type(df.index)}")        print(df.head())                if df.empty:            print("没有获取到新数据")            return                # ...数据处理与保存逻辑...

这是整个项目的核心函数，负责实际获取ETF每日价格数据。函数首先调用get_last_refreshed_date确定是否有现存数据文件。如果有，则计算出应获取数据的起始日期（即上次结束日期的下一天）；如果没有，则使用一个未来日期作为起始点（实际应用中应改为过去某日期）。

接下来，函数构造结束日期为当前日期，形成完整的查询时间范围。然后调用akshare提供的fund_etf_hist_sina方法获取指定ETF的历史数据。该方法返回一个DataFrame对象，包含了开盘价、最高价、最低价、收盘价和成交量等信息。

数据合并与更新逻辑

if last_refreshed_date:    with open(filename, 'r', encoding='utf-8') as f:        existing_data = json.load(f)        new_time_series = {}    for i, row in df.iterrows():        date_str = str(row['date'])        new_time_series[date_str] = {            "1. open": str(row['open']),            "2. high": str(row['high']),            "3. low": str(row['low']),            "4. close": str(row['close']),            "5. volume": str(row['volume'])        }        existing_time_series = existing_data.get("Time Series (Daily)", {})    existing_time_series.update(new_time_series)        data = existing_data    data["Meta Data"]["3. Last Refreshed"] = datetime.now().strftime("%Y-%m-%d")    data["Time Series (Daily)"] = existing_time_serieselse:    data = {        "Meta Data": {            "1. Information": "Daily Prices (open, high, low, close) and Volumes",            "2. Symbol": SYMBOL,            "3. Last Refreshed": datetime.now().strftime("%Y-%m-%d"),            "4. Output Size": "Full",            "5. Time Zone": "Asia/Shanghai"        }    }

这段代码展示了如何处理新旧数据的合并问题。当存在历史数据时，程序会先加载原有JSON文件中的内容，然后将新获取的数据逐条添加到时间序列字典中。这里使用了字典的update方法来实现高效合并，新数据会自动覆盖旧数据中相同日期的记录。

对于元数据部分，程序保留了原有的所有字段，仅更新了"Last Refreshed"时间为当前日期。这样做既保持了数据的连续性，又准确反映了最后一次更新的时间戳。

如果是第一次获取某只ETF的数据（即不存在历史文件的情况），程序会创建一个新的数据结构模板。这个模板遵循了Alpha Vantage API返回数据的格式规范，虽然本项目并未直接调用该API，但采用其数据结构有利于将来可能的迁移或兼容需求。

最终数据处理与保存

time_series = {}last_date = Nonefor i, row in df.iterrows():    date_str = str(row['date'])    time_series[date_str] = {        "1. open": str(row['open']),        "2. high": str(row['high']),        "3. low": str(row['low']),        "4. close": str(row['close']),        "5. volume": str(row['volume'])    }    last_date = date_strdata["Time Series (Daily)"] = time_seriesif last_date:    data["Meta Data"]["3. Last Refreshed"] = last_dateelse:    data["Meta Data"]["3. Last Refreshed"] = datetime.now().strftime("%Y-%m-%d")output_file = f'./daily_prices_{SYMBOL}.json'with open(output_file, 'w', encoding='utf-8') as f:    json.dump(data, f, ensure_ascii=False, indent=4)print(f"数据已保存到 {output_file}")

这部分代码完成了最后的数据处理工作。它遍历刚刚获取的DataFrame中的每一行数据，将其转换为特定的字典格式并存入time_series字典中。每条记录包含五个关键字段：开盘价、最高价、最低价、收盘价和成交量，分别标记为"1. open"到"5. volume"。

在处理完所有数据后，程序更新主数据结构的"Time Series (Daily)"键对应的值为刚构建好的时间序列字典。接着更新元数据中的"Last Refreshed"字段，优先使用最后一条数据的日期，若没有有效数据则使用当前日期。

最后一步是将整理好的数据写入本地文件系统。输出文件名为daily_prices_{SYMBOL}.json，位于程序运行目录下。json.dump函数负责将Python字典序列化为JSON文本，其中ensure_ascii=False参数保证了中文字符能正确显示而非转义序列，indent=4则使生成的JSON文件具有良好的缩进格式，提高了可读性，合并后的数据如下：

在这里插入图片描述

运行主程序 ：

调用mcp的服务去获取本地行情数据，构造提示词，调用ai得到交易指令。

python main.py
```![](http://cdn.zhipoai.cn/45c9daeb.jpg)

在这里插入图片描述

#### 运行web服务

```plaintext
cd docspython -m http.server 8000

访问 http://localhost:8000，可以看到qwen的收益曲线，投资组合分析中可以看到分别的持仓：

可以看到最近的交易操作，与大模型间的对话过程：

目前只获取了日线级别的数据，后面可补充周线、月线及分钟级数据会更有利于深入分析。还可以考虑引入更多类型的金融产品，比如LOF基金、分级基金甚至个别蓝筹股的数据，从而扩大研究范围，同时整理不同周期的技术指标，如MACD，RSI等，为大模型提供更丰富的数据支撑。

​最后

我在一线科技企业深耕十二载，见证过太多因技术卡位而跃迁的案例。那些率先拥抱 AI 的同事，早已在效率与薪资上形成代际优势，我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。

我整理出这套 AI 大模型突围资料包：

• ✅AI大模型学习路线图
• ✅Agent行业报告
• ✅100集大模型视频教程
• ✅大模型书籍PDF
• ✅DeepSeek教程
• ✅AI产品经理入门资料

如果你也想通过学大模型技术去帮助自己升职和加薪，可以扫描下方链接👇👇
​​

为什么说现在普通人就业/升职加薪的首选是AI大模型？

人工智能技术的爆发式增长，正以不可逆转之势重塑就业市场版图。从DeepSeek等国产大模型引发的科技圈热议，到全国两会关于AI产业发展的政策聚焦，再到招聘会上排起的长队，AI的热度已从技术领域渗透到就业市场的每一个角落。

智联招聘的最新数据给出了最直观的印证：2025年2月，AI领域求职人数同比增幅突破200% ，远超其他行业平均水平；整个人工智能行业的求职增速达到33.4%，位居各行业榜首，其中人工智能工程师岗位的求职热度更是飙升69.6%。

AI产业的快速扩张，也让人才供需矛盾愈发突出。麦肯锡报告明确预测，到2030年中国AI专业人才需求将达600万人，人才缺口可能高达400万人，这一缺口不仅存在于核心技术领域，更蔓延至产业应用的各个环节。

​​

资料包有什么？

①从入门到精通的全套视频教程

包含提示词工程、RAG、Agent等技术点

② AI大模型学习路线图（还有视频解说）

全过程AI大模型学习路线

③学习电子书籍和技术文档

市面上的大模型书籍确实太多了，这些是我精选出来的

④各大厂大模型面试题目详解

⑤ 这些资料真的有用吗?

这份资料由我和鲁为民博士共同整理，鲁为民博士先后获得了北京清华大学学士和美国加州理工学院博士学位，在包括IEEE Transactions等学术期刊和诸多国际会议上发表了超过50篇学术论文、取得了多项美国和中国发明专利，同时还斩获了吴文俊人工智能科学技术奖。目前我正在和鲁博士共同进行人工智能的研究。

所有的视频教程由智泊AI老师录制，且资料与智泊AI共享，相互补充。这份学习大礼包应该算是现在最全面的大模型学习资料了。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。

智泊AI始终秉持着“让每个人平等享受到优质教育资源”的育人理念‌，通过动态追踪大模型开发、数据标注伦理等前沿技术趋势‌，构建起"前沿课程+智能实训+精准就业"的高效培养体系。

课堂上不光教理论，还带着学员做了十多个真实项目。学员要亲自上手搞数据清洗、模型调优这些硬核操作，把课本知识变成真本事‌！

​​​​

如果说你是以下人群中的其中一类，都可以来智泊AI学习人工智能，找到高薪工作，一次小小的“投资”换来的是终身受益！

应届毕业生‌：无工作经验但想要系统学习AI大模型技术，期待通过实战项目掌握核心技术。

零基础转型‌：非技术背景但关注AI应用场景，计划通过低代码工具实现“AI+行业”跨界‌。

业务赋能 ‌突破瓶颈：传统开发者（Java/前端等）学习Transformer架构与LangChain框架，向AI全栈工程师转型‌。

👉获取方式：

😝有需要的小伙伴，可以保存图片到wx扫描二v码免费领取【保证100%免费】🆓**

​