第02章：A股行情监控推送工具

A股市场有1.9亿户个人投资者账户。哪怕只有0.001%的人愿意为一个精准的提醒工具付¥30/月，那也是57万用户。你不需要千分之一，你只需要一千个。

需求是什么

散户最常见的痛点：

“XX股票涨到XX价提醒我”——价格突破提醒
“XX股票放量异常提醒我”——成交量异常监控
“某个行业板块跌超X%提醒我”——板块波动监控
“某股票被机构大单买入提醒我”——资金流向监控

这些需求，通达信可以做，但操作复杂，通知不是微信；专业量化平台有，但价格贵（几千/月）；结果就是：大量散户宁愿每天盯着手机，也不知道有更方便的方式。

你的机会：一个月¥30-99，帮他们实现这件事，推送到微信。

技术实现框架

数据来源：新浪财经、东方财富提供的公开行情API（无需注册，直接调用）

# stock_monitor.py
import requests
import json
import time
import sqlite3
from datetime import datetime

def get_stock_price(stock_code: str) -> dict:
    """
    获取股票实时价格
    stock_code: 股票代码，如 "sh600036"（招商银行）或 "sz000001"（平安银行）
    """
    url = f"http://hq.sinajs.cn/list={stock_code}"
    headers = {
        "Referer": "https://finance.sina.com.cn",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
    }
    
    try:
        response = requests.get(url, headers=headers, timeout=5)
        response.encoding = "gbk"
        
        # 解析数据格式
        data_str = response.text.split('"')[1]
        if not data_str:
            return None
        
        fields = data_str.split(",")
        return {
            "name": fields[0],
            "open": float(fields[1]),
            "close_yesterday": float(fields[2]),
            "current": float(fields[3]),
            "high": float(fields[4]),
            "low": float(fields[5]),
            "volume": int(fields[8]),  # 成交量（手）
            "amount": float(fields[9]),  # 成交额（元）
            "time": fields[31],
        }
    except Exception as e:
        print(f"获取{stock_code}数据失败: {e}")
        return None


def check_alert_conditions(stock_data: dict, alerts: list) -> list:
    """
    检查是否满足提醒条件
    
    alerts格式：[
        {"type": "price_above", "value": 50.0, "note": "突破50元"},
        {"type": "price_below", "value": 45.0, "note": "跌破45元"},
        {"type": "volume_spike", "multiplier": 3.0, "note": "放量3倍"},
    ]
    """
    triggered = []
    current = stock_data["current"]
    
    for alert in alerts:
        if alert["type"] == "price_above" and current >= alert["value"]:
            triggered.append(f"价格突破 ¥{alert['value']}（当前：¥{current}）")
        
        elif alert["type"] == "price_below" and current <= alert["value"]:
            triggered.append(f"价格跌破 ¥{alert['value']}（当前：¥{current}）")
        
        elif alert["type"] == "volume_spike":
            # 简化：与前一日成交量比较（需要数据库存储历史均量）
            pass
    
    return triggered


def send_wechat_notify(message: str, webhook_url: str):
    """
    通过企业微信机器人发送通知
    webhook_url: 企业微信机器人的Webhook地址
    """
    payload = {
        "msgtype": "text",
        "text": {"content": message}
    }
    response = requests.post(webhook_url, json=payload)
    return response.status_code == 200


def monitor_loop(user_configs: list, interval_seconds: int = 60):
    """
    主监控循环
    
    user_configs格式：[
        {
            "user_id": "user_001",
            "webhook": "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx",
            "stocks": [
                {
                    "code": "sh600036",
                    "alerts": [
                        {"type": "price_above", "value": 50.0}
                    ]
                }
            ]
        }
    ]
    """
    while True:
        for user in user_configs:
            for stock_config in user["stocks"]:
                data = get_stock_price(stock_config["code"])
                if not data:
                    continue
                
                triggered = check_alert_conditions(data, stock_config["alerts"])
                
                for alert_msg in triggered:
                    message = f"📊 股票提醒\n股票：{data['name']}（{stock_config['code']}）\n{alert_msg}\n时间：{data['time']}"
                    send_wechat_notify(message, user["webhook"])
                    print(f"已发送提醒给 {user['user_id']}: {alert_msg}")
        
        time.sleep(interval_seconds)

用户配置管理

每个用户的提醒配置，存在SQLite中：

def init_db():
    conn = sqlite3.connect("stock_alerts.db")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS user_alerts (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id TEXT NOT NULL,
            stock_code TEXT NOT NULL,
            alert_type TEXT NOT NULL,  -- price_above/price_below/volume_spike
            alert_value REAL,
            webhook_url TEXT NOT NULL,
            is_active INTEGER DEFAULT 1,
            created_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
    """)
    conn.commit()
    return conn

商业化方案

免费层：最多监控2支股票，每天最多推送5条提醒 付费层（¥30/月）：最多监控10支股票，不限推送 高级层（¥99/月）：不限股票，支持自定义复杂条件

用户获取渠道：

股票相关微信群/QQ群
雪球、东方财富股吧的活跃用户
抖音/B站财经账号下的评论区

本章小结

A股监控是最容易入门的数据服务方向：数据来源公开，用户需求清晰
核心代码框架：定时拉取行情 → 判断条件 → 企业微信推送
定价：¥30-99/月，从朋友圈和股票社群开始获客

→ 第03章：小红书达人投放数据雷达