前言：打破指标钝化焦虑，让 AI 替你疯狂打工
传统的量化选股往往面临一个致命痛点：单纯依赖技术面指标（如 RSI、均线、突破等）极易产生假信号和指标钝化。 如果全市场几千只股票挨个去人工复盘，不仅耗时耗力，还会错失最佳建仓时机。
今天为你带来一套完全开源、高并发的轻量级量化策略扫描器系统。它不同于传统的“只报信号”的机器人，而是一个真正的 AI 选股 Agent。系统通过多线程并发调用行情接口，一旦命中你自定义的技术策略，立刻触发 Gemini 1.5 Flash 进行二次逻辑审计和打分，告别盲目跟从指标。最后，通过 PushDeer 一键将深度分析报告推送到你的手机端！
核心亮点
🚀 高并发扫描：基于
ThreadPoolExecutor多线程加速，配合tqdm进度条，成百上千只股票盘中 1 分钟扫完。🧠 AI 智能审计：过滤技术面杂音，由大模型进行基本面/形态综合复核，提供原因分析。
📱 零延迟推送：集成 PushDeer 服务，选股报告（代码、名称、技术信号、AI分析）直达手机通知栏。
⏱️ 完全自动化：默认每 30 分钟全自动循环轮询，适合盘中自动化运行。
💻 环境准备（磨刀不误砍柴工）
在开始部署前，请确保你的开发环境满足以下要求：
操作系统：Windows 10/11, macOS, Linux
Python 版本：Python 3.9 或以上版本
网络环境：由于需要调用 Gemini API，需要配置本地科学上网代理（默认支持 Clash 等工具的本地端口映射）。
推荐依赖组件
📂 项目推荐目录结构
为了让策略文件能够被系统动态加载，请在本地严格按照以下目录结构组织文件：
Plaintext
.
├── main.py # 主程序入口（核心控制流）
├── requirements.txt # 依赖项清单
└── strategies/ # 策略文件夹（存放你的选股武器库）
├── __init__.py
├── rsi_mom.py # 策略1：RSI动能策略
└── alpha_break.py # 策略2：阿尔法突破策略
⚠️ 策略文件编写规范：
任何放置在
strategies/目录下的策略文件，必须暴露出两个标准方法：
run(df: pd.DataFrame) -> bool: 接收行情 K 线数据，返回是否触发信号。
get_name() -> str: 返回该策略的易读名称（如 "RSI超买突破"）。
🛠️ 核心源码实现：main.py
请新建 main.py 并写入以下代码。（注意：为了你的资产安全，敏感 Key 已做脱敏处理，切勿直接将包含真实 Key 的代码上传至公共 GitHub 仓库！）
Python
import os, time, importlib.util, requests, json
import pandas as pd
from tqdm import tqdm
from concurrent.futures import ThreadPoolExecutor, as_completed
from threading import Lock
from google import genai
# --- 核心网络与接口配置 ---
proxy_addr = "http://127.0.0.1:7890" # 你的本地代理端口
os.environ['http_proxy'] = proxy_addr
os.environ['https_proxy'] = proxy_addr
os.environ['no_proxy'] = "localhost,127.0.0.1,192.168.11.48"
BASE_URL = "http://192.168.11.48:8080" # 本地行情API基地址
PUSH_DEER_KEY = os.getenv("PUSH_DEER_KEY") # 【安全避坑】从环境变量读取 PushDeer Key
GEMINI_KEY = os.getenv("GEMINI_API_KEY") # 【安全避坑】从环境变量读取 Gemini Key
STRATEGY_DIR = "./strategies"
TARGET_STRATEGIES = ["rsi_mom", "alpha_break"] # 指定运行的策略名
MAX_WORKERS = 20 # 并发线程数，根据机器和API承载力调整
REQUEST_TIMEOUT = (2, 5) # 连接/读取超时设置
# 初始化 Gemini 客户端
client = genai.Client(api_key=GEMINI_KEY)
def load_strategies():
"""动态加载策略模块"""
strategies = []
if not os.path.exists(STRATEGY_DIR):
os.makedirs(STRATEGY_DIR)
targets = TARGET_STRATEGIES or [
f[:-3] for f in os.listdir(STRATEGY_DIR)
if f.endswith(".py") and f != "__init__.py"
]
for name in targets:
path = os.path.join(STRATEGY_DIR, f"{name}.py" if not name.endswith(".py") else name)
if not os.path.exists(path):
print(f"⚠️ 策略文件未找到，已跳过: {path}")
continue
spec = importlib.util.spec_from_file_location(name, path)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
strategies.append(mod)
return strategies
def ai_audit(stock_code: str, tech_signal: str) -> dict:
"""利用 Gemini 1.5 Flash 进行智能二次复核"""
prompt = (
f"分析标的 {stock_code}，技术信号：{tech_signal}。"
'请按JSON格式返回: {"score": int, "reason": "str"}'
)
try:
resp = client.models.generate_content(model="gemini-1.5-flash", contents=prompt)
text = resp.text.replace("```json", "").replace("```", "").strip()
return json.loads(text)
except Exception as e:
return {"score": 0, "reason": f"AI分析失败: {e}"}
def scan_stock(stock: dict, strategies: list) -> tuple | None:
"""单只股票扫描线程任务。返回命中元组或 None"""
code = stock["code"]
name = stock.get("name", "未知")
try:
res = requests.get(
f"{BASE_URL}/api/kline?code={code}",
timeout=REQUEST_TIMEOUT,
headers={"Connection": "close"},
)
data = res.json().get("data", {}).get("List", [])
if not data:
return None
df = pd.DataFrame(data)
df.columns = [c.lower() for c in df.columns]
for s in strategies:
if s.run(df):
return (code, name, s.get_name()) # 命中策略，返回核心要素
except Exception:
pass # 单只票异常静默跳过，不影响大盘扫描
return None
def main():
strategies = load_strategies()
if not strategies:
print("❌ 没有可用策略，退出。")
return
try:
resp = requests.get(f"{BASE_URL}/api/codes", timeout=10)
stock_list = resp.json()["data"]["codes"]
except Exception as e:
print(f"💥 获取股票池失败: {e}")
return
hits = []
lock = Lock()
# 绚丽的终端进度条配置
bar_fmt = "{l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}, {rate_fmt}]"
with tqdm(total=len(stock_list), desc="🚀 量化扫描中", unit="只", bar_format=bar_fmt) as pbar:
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
futures = {pool.submit(scan_stock, s, strategies): s for s in stock_list}
for fut in as_completed(futures):
result = fut.result()
if result:
with lock:
hits.append(result)
pbar.update(1)
# 聚合结果并调用 AI 审计，大幅节约 Token 消耗
if hits:
report_lines = []
print(f"🎯 本轮技术面命中 {len(hits)} 只，开始接入 AI 智能审计...")
for code, name, signal in hits:
audit = ai_audit(code, signal)
report_lines.append(
f"【{name}】({code})\n技术面：{signal}\nAI分析：{audit['reason']}"
)
try:
requests.get(
"https://api2.pushdeer.com/message/push",
params={"pushkey": PUSH_DEER_KEY, "text": "🔔 AI量化掘金报告",
"desp": "\n\n".join(report_lines)},
proxies={"http": None, "https": None},
timeout=10,
)
print(f"✅ 报告推送完成，命中 {len(hits)} 只标的。")
except Exception as e:
print(f"⚠️ 推送失败: {e}")
else:
print("📭 本轮扫描结束，无技术面命中信号。")
if __name__ == "__main__":
while True:
main()
print("😴 等待 30 分钟后进行下一轮扫描...")
time.sleep(1800)
🎯 实测体验：AI 疯狂打工效果展示
运行程序后，终端会打印出丝滑的进度条：
Bash
🚀 量化扫描中: 100%|█████████████████████████| 3500/3500 [00:42<00:00, 83.21只/s]
🎯 本轮技术面命中 2 只，开始接入 AI 智能审计...
✅ 报告推送完成，命中 2 只标的。
打开手机 PushDeer App，你将收到如下格式的实时通知：
🔔 AI量化掘金报告
【腾讯控股】(00700)
技术面：rsi_mom (RSI动能超买突破)
AI分析：该标的在技术面突破的同时，近期伴随大单资金持续流入。AI 审计得分 85 分，形态处于健康的上升通道中，但需注意量能是否能持续放大，防范冲高回落风险。
🔒 核心避坑与 GitHub 上传安全提示
如果你准备将这个项目传到 GitHub 供大家学习或自己做备份，请牢记以下“保姆级”避坑指南：
绝对不要硬编码秘钥：
原始代码中包含
api_key="AIzaSy..."和PUSH_DEER_KEY="PDU4..."。一旦你直接git push，GitHub 的自动化安全扫描机制会在 5秒内 识别并通知 Google，你的 API Key 将会瞬间被官方封禁！正确利用环境变量：
本教程代码中已将 Key 改为了
os.getenv()形式。运行前只需在系统终端执行：Bash
export GEMINI_API_KEY="你的真实Gemini密钥" export PUSH_DEER_KEY="你的真实PushDeer密钥"配置文件过滤
.gitignore：在根目录下建立
.gitignore文件，写上：Plaintext
__pycache__/ .env .idea/ *.pyc
⚖️ 客观分析（不吹不黑）
优势 🌟
Token 消耗极低：系统采用“先技术后 AI”的过滤机制。全市场几千只票，经过多线程技术面筛选后通常只剩下寥寥数只，此时再调用 Gemini，每天运行也花不了几美分。
解耦设计：策略文件支持热加载，想加新策略直接往
strategies/丢一个文件即可，不需要改动主循环代码。
缺点与局限 ⚠️
网络依赖度高：国内环境运行对代理的稳定性有一定要求，如果代理断开，Gemini 审计部分会直接返回失败。
行情源并发限制：
MAX_WORKERS设为 20。如果你的本地行情服务器或券商 API 有频控，调得太高可能会被封禁 IP。
✍️ 总结
通过将多线程高效扫描与 Gemini AI 智能逻辑审计相结合，我们用不到 150 行代码，就搭建起了一套非常硬核盘中无人值守盯盘系统。快去为你自己配置一套，享受 AI 替你打工、实时推送掘金信号的爽感吧！
如果你觉得这个教程对你有帮助，欢迎在 GitHub 点个 Star！🌟
免责声明：本工具及相关策略仅用于技术交流与自动化工具开发，不构成任何实质性投资建议。股市有风险，量化需谨慎。
评论区