我用 Python 做了一个股票监控面板,跌到目标价自动喊我加仓

盯盘这件事

炒股的人有个毛病，总想看盘。早上开盘瞄一眼，中午吃饭瞄一眼，下午快收盘又瞄一眼。看的时候也不干什么，就是盯着那个数字跳。涨了两分钱高兴一下，跌了三毛心想想要不要割肉。

其实我给自己定过规矩：某只股票跌到某个价位才加仓，涨到某个价位就卖掉。但到了那个价位的时候，我往往不在看盘。等晚上复盘才发现，哦，今天跌到了我设的目标价，但没买。

手动盯盘这件事，浪费时间，还容易情绪化操作。所以我写了个小工具，让程序帮我盯。

这个工具能干什么

简单说，三件事：

1. 实时获取行情：每隔几秒拉取一次A股实时数据

2. 条件监控：你提前设好条件（跌到XX元、涨幅超过5%、成交量突破XX），满足条件就触发告警

3. 推送通知：触发告警后，把消息推到你手机上

支持同时监控多只股票，每只股票可以设多个条件。

技术选型

• akshare：A股数据接口，免费，数据全，不用申请token

• rich：终端美化库，在命令行里画表格和进度条

• requests：推送通知用（Server酱或企业微信）

不需要数据库，配置文件用JSON。

环境准备

pip install akshare rich requests

akshare是个宝藏库，A股、港股、美股、基金、期货、期权都能查到。我们只用它获取实时行情。

配置文件

创建一个 watch_config.json，把你关注的股票和告警条件写进去：

{"stocks": [    {"code": "sh600519","name": "贵州茅台","alerts": [        {"type": "price_below", "value": 1500, "msg": "茅台跌破1500，考虑加仓"},        {"type": "price_above", "value": 1800, "msg": "茅台涨过1800，考虑止盈"},        {"type": "change_pct_below", "value": -3, "msg": "茅台单日跌幅超3%"}      ]    },    {"code": "sz000858","name": "五粮液","alerts": [        {"type": "price_below", "value": 130, "msg": "五粮液跌破130，到加仓位了"},        {"type": "volume_above", "value": 50000000, "msg": "五粮液成交量放大到5000万手以上"}      ]    },    {"code": "sz300750","name": "宁德时代","alerts": [        {"type": "price_below", "value": 180, "msg": "宁德时代跌破180"},        {"type": "change_pct_above", "value": 5, "msg": "宁德时代单日涨幅超5%，注意追高风险"}      ]    }  ],"check_interval": 10,"notify_method": "console"}

说明一下各字段：

• code：股票代码，sh开头是上海，sz开头是深圳

• alerts.type 支持的条件类型：

• price_below：价格低于某值

• price_above：价格高于某值

• change_pct_below：涨跌幅低于某值（比如-3就是跌超3%）

• change_pct_above：涨跌幅高于某值

• volume_above：成交量大于某值

• check_interval：检查间隔，单位秒

• notify_method：通知方式，console 只在终端打印，后面会加推送

完整代码

"""股票监控面板实时获取A股行情，满足条件自动告警依赖安装：pip install akshare rich requests"""import jsonimport timefrom datetime import datetimefrom pathlib import Pathimport akshare as akfrom rich.console import Consolefrom rich.table import Tablefrom rich.live import Livefrom rich.panel import Panelfrom rich.layout import Layoutfrom rich.text import Textconsole = Console()# ============ 加载配置 ============CONFIG_FILE = Path("watch_config.json")def load_config():"""读取监控配置"""if not CONFIG_FILE.exists():console.print("[red]配置文件 watch_config.json 不存在[/red]")console.print("请先创建配置文件，参考文章中的示例")return Nonewith open(CONFIG_FILE, "r", encoding="utf-8") as f:return json.load(f)# ============ 行情获取 ============def get_realtime_quote(stock_code):"""    获取单只股票的实时行情    stock_code 格式：sh600519 或 sz000858    返回 dict，包含 price, change_pct, volume 等    """try:# akshare 的实时行情接口# 用 spot 接口获取全市场，然后筛选df = ak.stock_zh_a_spot_em()# 从代码中提取纯数字部分pure_code = stock_code[2:]# 找到对应行row = df[df["代码"] == pure_code]if row.empty:return Nonerow = row.iloc[0]return {"code": stock_code,"name": row["名称"],"price": float(row["最新价"]) ifrow["最新价"] != "-"else0,"change": float(row["涨跌额"]) ifrow["涨跌额"] != "-"else0,"change_pct": float(row["涨跌幅"]) ifrow["涨跌幅"] != "-"else0,"volume": float(row["成交量"]) ifrow["成交量"] != "-"else0,"amount": float(row["成交额"]) ifrow["成交额"] != "-"else0,"high": float(row["最高"]) ifrow["最高"] != "-"else0,"low": float(row["最低"]) ifrow["最低"] != "-"else0,"open": float(row["今开"]) ifrow["今开"] != "-"else0,"prev_close": float(row["昨收"]) ifrow["昨收"] != "-"else0,        }except Exception as e:console.print(f"[yellow]获取 {stock_code} 行情失败: {e}[/yellow]")return Nonedef get_all_quotes(stock_codes):"""批量获取所有股票行情（一次API调用）"""try:df = ak.stock_zh_a_spot_em()results = {}for code in stock_codes:pure_code = code[2:]row = df[df["代码"] == pure_code]if not row.empty:row = row.iloc[0]results[code] = {"code": code,"name": row["名称"],"price": float(row["最新价"]) ifrow["最新价"] != "-"else0,"change": float(row["涨跌额"]) ifrow["涨跌额"] != "-"else0,"change_pct": float(row["涨跌幅"]) ifrow["涨跌幅"] != "-"else0,"volume": float(row["成交量"]) ifrow["成交量"] != "-"else0,"amount": float(row["成交额"]) ifrow["成交额"] != "-"else0,"high": float(row["最高"]) ifrow["最高"] != "-"else0,"low": float(row["最低"]) ifrow["最低"] != "-"else0,"open": float(row["今开"]) ifrow["今开"] != "-"else0,"prev_close": float(row["昨收"]) ifrow["昨收"] != "-"else0,                }return resultsexcept Exception as e:console.print(f"[yellow]获取行情失败: {e}[/yellow]")return {}# ============ 告警判断 ============# 记录已触发的告警，避免重复通知triggered_alerts = set()def check_alerts(stock_config, quote):"""检查某只股票是否触发了告警条件"""if not quote:return []alerts_fired = []for alert in stock_config.get("alerts", []):alert_type = alert["type"]value = alert["value"]msg = alert["msg"]triggered = Falseif alert_type == "price_below" and quote["price"] < value:triggered = Trueelif alert_type == "price_above" and quote["price"] > value:triggered = Trueelif alert_type == "change_pct_below" and quote["change_pct"] < value:triggered = Trueelif alert_type == "change_pct_above" and quote["change_pct"] > value:triggered = Trueelif alert_type == "volume_above" and quote["volume"] > value:triggered = Trueif triggered:# 用 股票代码+条件类型+条件值 作为唯一标识，防止重复告警alert_key = f"{quote['code']}_{alert_type}_{value}"if alert_key not in triggered_alerts:triggered_alerts.add(alert_key)alerts_fired.append({"stock": quote["name"],"price": quote["price"],"msg": msg,"time": datetime.now().strftime("%H:%M:%S"),                })return alerts_fired# ============ 通知推送 ============def notify_console(alert):"""终端打印告警（默认方式）"""console.print(Panel(f"[bold red]{alert['stock']}[/bold red]\n"f"当前价格: {alert['price']}\n"f"{alert['msg']}\n"f"时间: {alert['time']}",title="!! 告警触发 !!",border_style="red",    ))def notify_serverchan(alert, send_key):"""通过 Server酱 推送到微信"""import requestsurl = f"https://sctapi.ftqq.com/{send_key}.send"data = {"title": f"股票告警: {alert['stock']}","desp": f"当前价格: {alert['price']}\n\n{alert['msg']}\n\n时间: {alert['time']}",    }try:resp = requests.post(url, data=data, timeout=10)if resp.json().get("code") == 0:console.print(f"[green]Server酱推送成功[/green]")else:console.print(f"[yellow]Server酱推送失败: {resp.text}[/yellow]")except Exception as e:console.print(f"[yellow]Server酱推送异常: {e}[/yellow]")def send_notification(alert, method, **kwargs):"""根据配置选择通知方式"""if method == "console":notify_console(alert)elif method == "serverchan":notify_serverchan(alert, kwargs.get("send_key", ""))# 你可以在这里加更多推送方式# ============ 面板渲染 ============def build_dashboard_table(quotes, stock_configs):"""构建终端行情面板表格"""table = Table(title="实时行情监控", show_header=True, header_style="bold cyan")table.add_column("股票", style="bold", width=12)table.add_column("现价", justify="right", width=10)table.add_column("涨跌额", justify="right", width=10)table.add_column("涨跌幅", justify="right", width=10)table.add_column("最高", justify="right", width=10)table.add_column("最低", justify="right", width=10)table.add_column("成交量(手)", justify="right", width=14)table.add_column("告警数", justify="center", width=8)for config in stock_configs:code = config["code"]q = quotes.get(code)if not q:table.add_row(config["name"], "--", "--", "--", "--", "--", "--", "0")continue# 涨跌幅颜色：红涨绿跌pct_str = f"{q['change_pct']:.2f}%"pct_style = "red"ifq["change_pct"] >= 0 else "green"# 统计该股票的告警条件数alert_count = len(config.get("alerts", []))table.add_row(q["name"],f"{q['price']:.2f}",f"{q['change']:+.2f}",Text(pct_str, style=pct_style),f"{q['high']:.2f}",f"{q['low']:.2f}",f"{q['volume']:,.0f}",str(alert_count),        )return table# ============ 主循环 ============def main():config = load_config()if not config:returnstock_configs = config["stocks"]interval = config.get("check_interval", 10)notify_method = config.get("notify_method", "console")stock_codes = [s["code"] forsinstock_configs]console.print(Panel(f"监控股票: {', '.join(s['name'] for s in stock_configs)}\n"f"刷新间隔: {interval}秒\n"f"通知方式: {notify_method}\n"f"按 Ctrl+C 退出",title="股票监控面板启动",border_style="blue",    ))try:with Live(console=console, refresh_per_second=1) as live:while True:# 获取行情quotes = get_all_quotes(stock_codes)# 构建面板table = build_dashboard_table(quotes, stock_configs)live.update(table)# 检查告警for config_item in stock_configs:code = config_item["code"]quote = quotes.get(code)alerts = check_alerts(config_item, quote)for alert in alerts:send_notification(alert, notify_method)time.sleep(interval)except KeyboardInterrupt:console.print("\n[yellow]监控已停止[/yellow]")if __name__ == "__main__":main()

运行效果

python stock_monitor.py

终端会出现一个实时刷新的表格：

┌──────────────────────────────────────────────────────────────────────┐│                        实时行情监控                                   │├────────────┬──────────┬──────────┬──────────┬──────────┬──────────┬────────────────┬──────────┤│ 股票       │     现价 │    涨跌额 │    涨跌幅 │     最高 │     最低 │   成交量(手)   │ 告警数   │├────────────┼──────────┼──────────┼──────────┼──────────┼──────────┼────────────────┼──────────┤│ 贵州茅台   │ 1523.00  │   +12.50 │  +0.83%  │ 1530.00  │ 1508.00  │     32,456     │    3     ││ 五粮液     │  128.50  │    -2.30 │  -1.76%  │  131.20  │  127.80  │    187,234     │    2     ││ 宁德时代   │  185.60  │    +6.80 │  +3.80%  │  187.40  │  178.50  │    456,789     │    2     │└────────────┴──────────┴──────────┴──────────┴──────────┴──────────┴────────────────┴──────────┘

当五粮液的当前价跌到128.50，低于你设的130目标价，终端会弹出告警面板：

╭────────── !! 告警触发 !! ──────────╮│ 五粮液                              ││ 当前价格: 128.5                     ││ 五粮液跌破130，到加仓位了            ││ 时间: 14:23:15                      │╰────────────────────────────────────╯

如果你配置了Server酱，这条消息还会推送到你的微信。

几个实用技巧

避免重复告警。代码里用了一个 triggered_alerts 集合来记录已触发的条件。同一个条件在一个运行周期内只会通知一次。第二天重新运行脚本会重置。

只在交易时间监控。 你可以在主循环里加个判断，9:30到15:00之间才拉数据，其他时间休眠。这样不浪费资源：

from datetime import datetimenow = datetime.now()hour_min = now.hour*100+now.minute# 9:30 到 11:30，或 13:00 到 15:00is_trading_time = (930<= hour_min<= 1130) or (1300<= hour_min<= 1500)

Server酱怎么用。 去 sct.ftqq.com 注册，用微信扫码登录，拿到你的SendKey。把它填到通知函数里就行。免费版每天5条推送，炒股够用了。

更多告警条件。 你可以自己扩展 check_alerts 函数。比如加一个“换手率超过某值”、“成交额突破某值”、“涨停/跌停”。只要 akshare 返回的数据里有这个字段，就能加。

免责声明

这个工具只是一个行情监控和价格提醒工具，不构成任何投资建议。股市有风险，买卖决策请自行判断。

工具能帮你不错过目标价位，但买不买、卖不卖，还是你自己说了算。

最后

盯盘这件事，本质上是一种焦虑。你怕错过机会，怕亏损扩大，怕该买的时候没买。

但程序没有焦虑。它只看数字，到了就喊你，没到就等着。

把盯盘的时间省出来，写写代码，看看书，或者好好吃顿饭。让机器去做机器擅长的事。

下一篇想写什么，评论区告诉我。

https://ima.qq.com/wiki/?shareId=f2628818f0874da17b71ffa0e5e8408114e7dbad46f1745bbd1cc1365277631c