在量化投资与个人资产管理场景中，实时掌握标的价格变动并及时响应，是提升决策效率的核心需求。传统人工盯盘方式不仅耗费精力，还容易错过关键交易窗口。通过Python实现自动化行情采集与预警推送，能够7×24小时监控市场异动，显著提升信息获取效率。
本文将从实战角度出发，详细讲解如何基于同花顺数据源搭建一套完整的价格预警系统，涵盖数据采集、规则引擎、多渠道通知三大核心模块，并给出可直接运行的核心代码实现。
一、系统整体架构设计
一套稳定可用的行情预警系统，需要在数据获取的实时性、规则判断的灵活性、通知推送的可靠性之间取得平衡。我们将系统拆分为四个独立模块，各模块通过标准化数据接口解耦，便于后续扩展维护。
数据采集层：负责从同花顺平台获取实时行情数据，支持轮询与订阅两种模式，内置异常重试与频率控制机制。
数据处理层：对原始行情数据进行清洗、格式化与指标计算，输出标准化的行情快照供规则引擎消费。
规则引擎层：支持多种预警条件配置，包括价格阈值、涨跌幅限制、技术指标触发等，采用状态机避免重复告警。
通知推送层：集成邮件、企业微信、钉钉等多种通知渠道，支持分级告警与静默期配置。
二、开发环境准备
2.1 依赖库选型
针对同花顺数据采集，推荐使用成熟的开源工具库而非从零编写采集逻辑。pywencai是专门对接同花顺问财平台的Python库，内部封装了参数加密与请求逻辑，能够大幅降低开发成本。同时搭配pandas做数据处理，APScheduler做定时调度。
核心依赖清单：
- pywencai：同花顺问财数据接口封装
- pandas：结构化数据处理
- APScheduler：定时任务调度
- smtplib：邮件通知发送
- python-dotenv：配置文件管理
2.2 环境安装
pip install pywencai pandas apscheduler python-dotenv
安装完成后，可通过简单调用验证环境是否正常。该库首次运行会自动下载浏览器驱动，用于模拟真实访问环境，绕过平台的前端校验机制。
三、同花顺行情数据采集实现
3.1 实时行情获取
同花顺平台的行情数据通过前端动态渲染，直接发起HTTP请求无法获取有效数据。pywencai底层通过无头浏览器执行页面脚本，等待数据加载完成后提取结构化结果，这也是目前最稳定的采集方式。
核心采集代码：
import pywencai
import pandas as pd
def get_realtime_quote(stock_codes):
"""获取指定股票的实时行情数据"""
query = f"{','.join(stock_codes)} 现价 涨跌幅 成交量 成交额"
result = pywencai.get(query=query, loop=True)
if result and len(result) > 0:
df = result[0]
return df[['股票代码', '股票简称', '现价', '涨跌幅', '成交量', '成交额']]
return pd.DataFrame()
调用时传入股票代码列表即可一次性获取多只标的的实时行情。返回结果为DataFrame格式，便于后续的数据处理与条件判断。
3.2 采集稳定性优化
金融数据采集对稳定性要求较高，单次请求失败可能导致预警漏报。实际部署时需要增加三层防护：
请求重试机制：网络波动或服务端限流时自动重试，采用指数退避策略，避免短时间内频繁请求触发防护机制。
频率控制：两次采集间隔不低于3秒，单批次采集标的数量控制在50只以内，降低对目标服务器的压力。
异常兜底：当主采集通道不可用时，自动切换至备用数据源，确保监控不中断。
3.3 历史K线数据补充
除实时行情外，预警系统往往需要结合历史走势进行判断。例如价格突破20日均线、成交量异常放大等条件，都需要历史K线数据支撑。
def get_kline_data(stock_code, period='daily', count=60):
"""获取历史K线数据"""
query = f"{stock_code} {period}K线 收盘价 成交量 开盘价 最高价 最低价"
result = pywencai.get(query=query)
return result[0] if result else pd.DataFrame()
获取K线数据后，可通过TA-Lib或pandas计算各类技术指标，支撑更复杂的预警规则。
四、价格预警规则引擎设计
4.1 核心预警规则类型
实际投资场景中，常用的预警规则可分为四大类，覆盖绝大多数盯盘需求：
价格阈值预警：最基础也最常用的类型，包括价格突破上轨、跌破下轨、达到目标价等场景。
涨跌幅预警：监控日内涨跌幅，当涨跌幅度超过设定阈值时触发告警，适合捕捉异动行情。
成交量预警：当成交量显著放大或萎缩时发出提示，常用于判断资金进出与趋势延续性。
技术指标预警：基于MA、MACD、RSI等技术指标的金叉死叉、超买超卖信号触发预警。
4.2 规则引擎实现
规则引擎采用配置化设计，所有预警规则通过JSON格式配置，无需修改代码即可调整监控策略。每条规则包含标的代码、规则类型、阈值参数、通知渠道等字段。
class AlertEngine:
def __init__(self, rules_config):
self.rules = rules_config
self.triggered = set() # 记录已触发的规则，避免重复告警
def check_price_threshold(self, quote, rule):
"""检查价格阈值规则"""
price = float(quote['现价'])
if rule['direction'] == 'above' and price >= rule['threshold']:
return True, f"价格突破{rule['threshold']}元，当前{price}元"
if rule['direction'] == 'below' and price <= rule['threshold']:
return True, f"价格跌破{rule['threshold']}元，当前{price}元"
return False, ""
为避免同一条件反复触发通知造成骚扰，引擎维护了已触发规则集合，规则从满足状态回到不满足状态后，才会重置告警状态，确保每次有效突破只推送一次通知。
4.3 规则配置示例
{
"stock_code": "300750",
"stock_name": "宁德时代",
"rules": [
{
"type": "price_above",
"threshold": 220.0,
"channel": "email",
"level": "normal"
},
{
"type": "price_below",
"threshold": 180.0,
"channel": "wechat",
"level": "urgent"
}
]
}
五、多渠道通知推送
预警信息的触达及时性直接影响系统价值。本文实现邮件与企业微信两种主流通知方式，可根据实际需求扩展钉钉、短信、电话等渠道。
5.1 邮件通知
邮件通知适合非紧急类告警，优点是免费、无发送频次限制，适合留存预警记录。使用SMTP协议发送，主流邮箱均支持。
import smtplib
from email.mime.text import MIMEText
from email.header import Header
def send_email_alert(subject, content, to_addr):
msg = MIMEText(content, 'plain', 'utf-8')
msg['From'] = Header('行情预警系统')
msg['To'] = Header(to_addr)
msg['Subject'] = Header(subject, 'utf-8')
with smtplib.SMTP_SSL('smtp.qq.com', 465) as server:
server.login('sender@qq.com', 'auth_code')
server.sendmail('sender@qq.com', [to_addr], msg.as_string())
5.2 企业微信/钉钉通知
企业微信与钉钉的机器人Webhook方式，推送延迟低、移动端提醒及时，适合紧急级别的预警。通过HTTP POST请求即可完成消息推送，接入成本极低。
import requests
import json
def send_wechat_alert(webhook_url, content):
payload = {
"msgtype": "text",
"text": {"content": f"【行情预警】\\n{content}"}
}
requests.post(webhook_url, data=json.dumps(payload),
headers={'Content-Type': 'application/json'})
六、系统调度与运行
使用APScheduler的BlockingScheduler调度器，在交易时段内按固定间隔执行采集与检测逻辑。A股交易时段为工作日的9:30-11:30和13:00-15:00，调度器可配置为仅在交易时间内运行。
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.interval import IntervalTrigger
import datetime
def is_trading_time():
"""判断当前是否为A股交易时间"""
now = datetime.datetime.now()
if now.weekday() >= 5:
return False
morning = (datetime.time(9, 30), datetime.time(11, 30))
afternoon = (datetime.time(13, 0), datetime.time(15, 0))
t = now.time()
return morning[0] <= t <= morning[1] or afternoon[0] <= t <= afternoon[1]
def monitor_task():
if not is_trading_time():
return
quotes = get_realtime_quote(stock_list)
engine.check_all_rules(quotes)
if __name__ == '__main__':
scheduler = BlockingScheduler()
scheduler.add_job(monitor_task, IntervalTrigger(seconds=10))
scheduler.start()
采集间隔可根据需求调整，一般个股监控设置为10-30秒即可满足大多数场景，过于频繁不仅加重服务器负载，也容易触发平台的访问限制。
七、部署与运维建议
7.1 服务化部署
生产环境建议将系统部署为后台服务，可使用systemd或supervisor进行进程管理，确保程序异常退出后能自动重启。同时将配置项与代码分离，敏感信息如邮箱授权码、Webhook地址等通过环境变量注入。
7.2 日志与监控
完善的日志是排查问题的关键。建议按天切割日志文件，记录每次采集的返回状态、规则匹配结果、通知发送情况。关键指标异常时可触发自监控告警，形成闭环。
7.3 性能优化建议
当监控标的数量较多时，可采用分批采集策略，每批处理固定数量的股票，批次之间增加间隔。数据处理与规则判断逻辑尽量向量化执行，充分利用pandas的批量计算能力，避免逐行循环带来的性能损耗。
合规与风险提示
本文所述技术方案仅用于个人学习与研究用途。数据采集行为应当遵守目标网站的服务条款与robots协议，控制请求频率，避免对服务器造成不必要的压力。
金融数据具有较高的商业价值，未经授权不得用于商业用途或二次分发。实际投资决策请以官方渠道数据为准，自动化预警系统仅作为辅助工具，不构成任何投资建议。系统运行过程中如遇数据接口变更或访问限制，请及时调整采集策略，尊重平台的数据权益。
