从零打造：利用 AI 模型实现自动化量化选股扫描-硅基核心 (Silicon Core)

本教程将带您深入实战，利用 Python 与 Google Gemini 大模型搭建一套自动化量化选股系统。我们将不仅涵盖基础的技术指标策略编写，还会引入 AI 辅助决策机制，让您的交易系统具备“机器之心”

系统架构总览图
- 描述：展示数据流动方向。
- 节点：API接口 -> Pandas数据处理 -> 策略模块 (Strategies) -> Gemini AI 审计 -> PushDeer 推送。
- 标签：
模块化策略逻辑图
- 描述：解释 load_strategies() 如何动态加载 bollinger.py、donchian.py 等插件。
- 标签：

系统核心流程解析

数据源部署：【点击直接】

配置层与环境准备：代码通过环境变量配置代理（proxy_addr），并明确排除推送域名，确保网络链路在复杂环境下的连通性。

策略解耦设计：

采用动态导入机制（importlib.util），将交易策略（布林带、唐奇安、华尔街趋势）独立为 .py 文件存放在 strategies 文件夹中。
这种设计允许用户在不修改 main.py 主程序的情况下，通过新增策略文件实现功能扩展。

import pandas as pd

def get_name(): return "华尔街趋势突破"

def run(df):
    df = df.rename(columns={'volume': 'vol', 'close': 'c'})
    if len(df) < 20: return False
    
    # 逻辑：MA20 趋势 + 量比 > 2 + 涨幅区间
    ma20 = df['c'].rolling(20).mean().iloc[-1]
    vol_ratio = df['vol'].iloc[-1] / df['vol'].rolling(5).mean().iloc[-2]
    change = (df['c'].iloc[-1] - df['c'].iloc[-2]) / df['c'].iloc[-2]
    
    return df['c'].iloc[-1] > ma20 and vol_ratio > 1.5 and 0.04 < change < 0.09

AI 智能审计流水线：

系统获取股票数据并转化为 pandas.DataFrame 格式，运行预设指标逻辑。
一旦触发策略信号，系统将标的代码与信号名称提交至 gemini-1.5-flash 模型。
AI 模型会返回 JSON 格式的评分与逻辑分析，最终通过 PushDeer 将筛选结果实时推送到移动端。

import pandas as pd
import requests
import google.generativeai as genai
import json

# 配置 Gemini
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel('gemini-1.5-flash')

# 1. 模拟数据获取 (从 tdx-api 转换而来)
def get_stock_data(symbol):
    # 此处调用 tdx-api 并转化为 DataFrame
    # 假设返回的数据包含：日期, 开盘, 最高, 最低, 收盘, 成交量
    data = {'date': ['2026-06-05'], 'close': [10.5], 'vol': [10000]}
    return pd.DataFrame(data)

# 2. 策略逻辑判断 (预设技术指标)
def check_strategy(df):
    # 简单示例：收盘价大于均线则触发
    if df['close'].iloc[-1] > 10.0:
        return True, "金叉信号"
    return False, None

# 3. AI 深度审计 (Gemini 推理)
def audit_with_ai(symbol, signal_name, data):
    prompt = f"""
    你是专业的量化审计分析师。标的: {symbol}，触发策略: {signal_name}。
    近几日行情数据: {data.to_json()}。
    请分析该信号的可靠性，给出 1-10 的评分，并用简短逻辑说明。
    返回格式必须为纯 JSON: {{"score": int, "analysis": "string"}}
    """
    response = model.generate_content(prompt)
    return json.loads(response.text)

# 4. 实时推送 (PushDeer)
def push_notification(content):
    url = "https://api2.pushdeer.com/message/push"
    params = {"pushkey": "YOUR_PUSHKEY", "text": "量化审计提醒", "desp": content}
    requests.get(url, params=params)

# 主循环
def main():
    symbol = "000001.SZ"
    df = get_stock_data(symbol)
    triggered, signal_name = check_strategy(df)
    
    if triggered:
        result = audit_with_ai(symbol, signal_name, df)
        if result['score'] > 7:  # 只有高分才推送
            msg = f"【{symbol}】{signal_name} 触发！\nAI 审计分: {result['score']}\n分析: {result['analysis']}"
            push_notification(msg)

if __name__ == "__main__":
    main()

异常熔断机制：
- 在扫描过程中，使用 timeout 参数限制网络请求，并结合 try-except 捕获异常，确保单只股票的数据延迟不会导致整个循环卡死，保证程序 24/7 运行。
- 工程代码【点击下载】

目录CONTENT

从零打造：利用 AI 模型实现自动化量化选股扫描

系统核心流程解析

评论区