A 股分时量能分析・实时量价中枢：逐笔资金 + 分时量价 + 背离共振 + 主力行为解析（完整二开代码）

一、核心定位

A 股分时量能分析（RealTime Volume-Price） 是逐分钟 / 逐笔级分时量价与资金行为解析引擎，专注日内量价同步 / 背离、主力真假动作、盘口资金流向、分时强弱、精准买卖点，对接 Tushare / 新浪 / 同花顺实时数据源，毫秒级刷新、全量指标、智能识别、可视化看板、告警推送，是日内交易、短线盯盘、主力盘口拆解的核心工具。

核心：量是因、价是果、分时是资金心电图
目标：识别真涨 / 假涨、真跌 / 假跌、主力洗盘 / 出货 / 吸筹

二、传统分时工具痛点

❌ 只看价格，无量能 / 资金验证，真假突破分不清
❌ 无逐笔大单 / 主力流向，看不到盘口真实资金
❌ 背离识别弱（顶背离 / 底背离），信号滞后
❌ 无量能形态：递增量 / 对倒量 / 洗盘量 / 恐慌量无法识别
❌ 无均价线强弱、支撑压力、量比、换手、振幅联动
❌ 无实时告警、截图、复盘日志、二开接口

三、核心功能（全维度分时量价）

（一）实时分时数据（1 分钟 / 500ms）

分时线：现价、均价线（VWAP）、昨收、今开、最高、最低
分时量柱：每分钟成交量、成交额、红绿柱（涨红跌绿）
逐笔成交：大单 / 超大单 / 中单 / 散户、买盘 / 卖盘、主动买入 / 卖出
量比 / 换手率 / 振幅 / 委比 / 委差

（二）量价关系引擎（核心）

价涨量增（健康多头）：价格上 + 量柱放大→主力真买，延续上涨
价涨量缩（顶背离・卖）：价格新高 + 量萎缩→虚涨、诱多、出货
价跌量增（空头强）：价格下 + 量放大→恐慌抛压、主力出货
价跌量缩（底背离・买）：价格新低 + 量萎缩→抛压竭尽、承接强、反弹
放量突破 / 破位：放量过均价 / 前高、放量跌破支撑

（三）分时量能形态识别（主力行为）

✅ 递增量（吸筹 / 蓄力）：量柱温和放大、阶梯上升→健康上涨
✅ 对倒量（诱多 / 出货）：单根巨量、瞬间萎缩→主力自买自卖
✅ 洗盘量（震荡洗盘）：价跌、量缩、不破均价→清洗浮筹
✅ 恐慌量（急跌见底）：急跌 + 巨量→恐慌盘出清、承接进场
✅ 缩量横盘（等待突破）：窄幅震荡、量低迷→变盘前兆

（四）主力资金流向（盘口拆解）

主力净流入：超大单 (>100 万)+ 大单 (20-100 万) 净额
资金强度：主力 / 散户资金占比、主动买 / 主动卖比例
盘口挂单：买一～买五、卖一～卖五、托单 / 压单、异常挂单
分时资金曲线：实时主力资金线、红绿柱、背离预警

（五）分时强弱 & 支撑压力

线上强、线下弱、震荡观望
- 股价持续在均价线上方：多头强、支撑有效
- 股价持续在均价线下方：空头强、压力有效
- 股价围绕均价缠绕：多空平衡、变盘等待
关键位：开盘价、昨收、均价、日内高低、前高 / 前低

（六）智能告警 & 信号

顶背离告警、底背离告警、放量拉升、放量跳水、突破均价、跌破均价、大单异动、对倒量
买卖点信号：背离反转、量价共振、突破确认、资金拐头

（七）可视化看板

主图：分时价格 + 均价线
副图 1：分时成交量柱
副图 2：主力资金流向
副图 3：量比 / 换手 / 振幅
实时数据面板、异动滚动、信号标记

四、完整二次开发代码（Python+FastAPI+ECharts，可直接部署）

1. 项目结构

plaintext

volume-price/
├── config/          # 配置
├── core/            # 数据、量价、资金、背离引擎
├── api/             # FastAPI接口
├── web/             # ECharts前端看板
├── utils/            # 日志、工具
├── index.py          # 入口
└── requirements.txt

2. requirements.txt

plaintext

fastapi
uvicorn
requests
pandas
numpy
akshare
tushare
python-multipart

3. 配置 config/settings.py

python

运行

# 数据源：新浪/同花顺/tushare
DATA_SOURCE = "sina"
TUSHARE_TOKEN = ""
# 刷新间隔（ms）
REFRESH_INTERVAL = 500
# 背离阈值
DIVERGENCE_THRESHOLD = 0.8
# 大单阈值（万）
BIG_ORDER = 20
SUPER_ORDER = 100

4. 数据引擎 core/data_engine.py

python

运行

import requests
import pandas as pd
import time
from config.settings import *

class MinuteDataEngine:
    def __init__(self):
        self.session = requests.Session()

    # 获取A股分时（新浪）
    def get_minute_data(self, code):
        """
        返回：time, price, volume, amount, avg_price
        """
        try:
            url = f"https://hq.sinajs.cn/list={code}"
            res = self.session.get(url, timeout=3)
            data = res.text.split(",")
            # 解析基础
            name = data[0].split("=")[1].replace('"','')
            open_p = float(data[1])
            last_close = float(data[2])
            high = float(data[4])
            low = float(data[5])
            # 分时明细（5日）
            url2 = f"https://money.finance.sina.com.cn/quotes_service/api/json_v2.php/CN_MarketData.getKLineData?symbol={code}&scale=1&ma=5&datalen=240"
            df = pd.read_json(url2)
            df['time'] = pd.to_datetime(df['day'])
            df['price'] = df['close']
            df['volume'] = df['volume']
            df['amount'] = df['amount']
            df['avg_price'] = df['amount'] / df['volume']
            return df
        except Exception as e:
            print("数据获取失败:", e)
            return None

    # 逐笔成交
    def get_tick_data(self, code):
        url = f"https://hq.sinajs.cn/list={code}_tick"
        res = self.session.get(url)
        return res.text

5. 量价核心引擎 core/volume_price.py

python

运行

import pandas as pd
import numpy as np

class VolumePriceAnalyzer:
    def __init__(self):
        self.data = None

    def set_data(self, df):
        self.data = df

    # 均价线VWAP
    def calc_vwap(self):
        if self.data is None:
            return None
        self.data['vwap'] = (self.data['amount'].cumsum() / self.data['volume'].cumsum())
        return self.data['vwap']

    # 量价同步/背离
    def check_divergence(self):
        """
        返回：顶背离、底背离、信号
        """
        if self.data is None or len(self.data) < 20:
            return {}
        # 价格高点/低点
        price_high = self.data['price'].rolling(5).max()
        price_low = self.data['price'].rolling(5).min()
        # 量能高点/低点
        vol_high = self.data['volume'].rolling(5).max()
        vol_low = self.data['volume'].rolling(5).min()

        # 顶背离：价新高、量降低
        top_div = (price_high.iloc[-1] > price_high.iloc[-5]) and (vol_high.iloc[-1] < vol_high.iloc[-5] * 0.8)
        # 底背离：价新低、量升高
        bot_div = (price_low.iloc[-1] < price_low.iloc[-5]) and (vol_low.iloc[-1] > vol_low.iloc[-5] * 1.2)

        return {
            "top_divergence": top_div,
            "bottom_divergence": bot_div,
            "signal": "SELL" if top_div else "BUY" if bot_div else "NEUTRAL"
        }

    # 量能形态识别
    def volume_pattern(self):
        if self.data is None:
            return "无数据"
        vol = self.data['volume'].tail(10)
        # 递增量
        if vol.is_monotonic_increasing and vol.iloc[-1] > vol.iloc[0] * 1.5:
            return "递增量（吸筹）"
        # 对倒量
        if vol.iloc[-1] > vol.mean() * 3 and vol.iloc[-2] < vol.mean():
            return "对倒量（诱多）"
        # 缩量
        if vol.iloc[-1] < vol.mean() * 0.5:
            return "缩量（观望）"
        return "正常量能"

    # 分时强弱
    def strength(self):
        if self.data is None:
            return "未知"
        last_price = self.data['price'].iloc[-1]
        vwap = self.data['vwap'].iloc[-1]
        if last_price > vwap * 1.01:
            return "强势（多头）"
        elif last_price < vwap * 0.99:
            return "弱势（空头）"
        else:
            return "震荡（平衡）"

6. 主力资金引擎 core/money_flow.py

python

运行

class MoneyFlowAnalyzer:
    def __init__(self):
        self.big_order = 20
        self.super_order = 100

    # 计算主力资金
    def calc_main_money(self, tick_data):
        """
        超大单、大单、净流入
        """
        super_buy = 0
        super_sell = 0
        big_buy = 0
        big_sell = 0
        # 解析逐笔
        lines = tick_data.split(";")
        for line in lines:
            if not line:
                continue
            parts = line.split(",")
            price = float(parts[1])
            volume = int(parts[2])
            direction = parts[3] # B/S
            amount = price * volume / 10000
            if amount >= self.super_order:
                if direction == "B":
                    super_buy += amount
                else:
                    super_sell += amount
            elif amount >= self.big_order:
                if direction == "B":
                    big_buy += amount
                else:
                    big_sell += amount
        net = (super_buy + big_buy) - (super_sell + big_sell)
        return {
            "super_buy": super_buy,
            "super_sell": super_sell,
            "big_buy": big_buy,
            "big_sell": big_sell,
            "net_inflow": net
        }

7. FastAPI 接口 api/main.py

python

运行

from fastapi import FastAPI
from core.data_engine import MinuteDataEngine
from core.volume_price import VolumePriceAnalyzer
from core.money_flow import MoneyFlowAnalyzer

app = FastAPI(title="A股分时量能分析")
data_engine = MinuteDataEngine()
vp_analyzer = VolumePriceAnalyzer()
money_analyzer = MoneyFlowAnalyzer()

@app.get("/minute/{code}")
def get_minute(code: str):
    df = data_engine.get_minute_data(code)
    if df is None:
        return {"code": code, "data": None}
    vp_analyzer.set_data(df)
    vp_analyzer.calc_vwap()
    div = vp_analyzer.check_divergence()
    pattern = vp_analyzer.volume_pattern()
    strength = vp_analyzer.strength()
    return {
        "code": code,
        "time": df['time'].tolist(),
        "price": df['price'].tolist(),
        "volume": df['volume'].tolist(),
        "vwap": df['vwap'].tolist(),
        "divergence": div,
        "pattern": pattern,
        "strength": strength
    }

@app.get("/moneyflow/{code}")
def get_moneyflow(code: str):
    tick = data_engine.get_tick_data(code)
    flow = money_analyzer.calc_main_money(tick)
    return flow

8. 前端看板 web/index.html（ECharts）

html

预览

<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>A股分时量能分析</title>
<script src="https://cdn.bootcdn.net/ajax/libs/echarts/5.4.3/echarts.min.js"></script>
</head>
<body>
<div id="main" style="width:1200px;height:600px;"></div>
<script>
// 自动刷新
setInterval(() => {
    fetch('/minute/600519.SH').then(r=>r.json()).then(data=>{
        // 渲染分时+量能
        let myChart = echarts.init(document.getElementById('main'));
        let option = {
            tooltip: {trigger: 'axis'},
            grid: [{height: '70%'}, {top: '75%', height: '25%'}],
            xAxis: [{data: data.time}, {data: data.time}],
            yAxis: [{scale:true}, {scale:true}],
            series: [
                {name:'价格', type:'line', data:data.price},
                {name:'均价', type:'line', data:data.vwap},
                {name:'成交量', type:'bar', data:data.volume, xAxisIndex:1, yAxisIndex:1}
            ]
        };
        myChart.setOption(option);
    })
}, 500);
</script>
</body>
</html>

9. 启动入口 index.py

python

运行

import uvicorn
from api.main import app

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

五、实战口诀（直接用）

线上多、线下空、缠绕不交易
价涨量增放心买，价涨量减快离场
价跌量减是见底，价跌量增继续跌
放量突破真突破，缩量突破是诱多
对倒量后不追高，洗盘量后低吸多

六、联动 OpenClaw 全生态

Stock Monitor：分时信号→告警→监控联动
Agent Browser：自动截图分时图、保存复盘
tushare-finance：历史分时回测、量价策略验证
飞书推送：异动 / 背离 / 信号实时推送

七、二开拓展方向

AI 分时预测：LSTM 预测未来 5-15 分钟量价
多股对比：同板块 / 同概念分时量价联动
盘口挂单解析：托单 / 压单 / 异常挂单识别
日内策略回测：量价背离 / 共振策略自动回测
WebSocket 实时推送：前端毫秒级无刷新