2.35万亿逼空行情：Python从零构建
情绪与资金流实时监控系统

2026年4月16日 | 真龙现身

4月16日，全A成交额2.35万亿，创业板指涨3.17%创近十一年新高。

作为一个写代码的工程师，那天我没有盯盘，而是让一套我自己写的监控系统告诉我市场发生了什么。这套系统用Python搭建，可以实时追踪情绪因子、资金流向和技术信号。收盘的时候，它给了我一幅清晰的图景，比任何朋友圈解读都准确。

现在市面上有无数炒股的App，告诉你"资金流入"、"主力净流入"这些概念。但你用过之后会发现，它们的数据往往存在明显的滞后，而且数据口径不统一，不同App的数据经常打架。

我解决这个问题的思路是：自己动手，用Python把数据源接进来，做自己的分析逻辑，输出自己的信号。这样我知道数据的来源是什么，我知道计算口径是什么，我也能根据市场变化随时调整。

这套监控系统，主要追踪三类数据：

1. 北向资金流
通过东方财富等数据源获取沪深股通的外资每日净流入数据。我的模型会把北向资金的日变化率做成一个时间序列，用来判断外资对A股的整体情绪。

2. 融资融券余额变化
融资余额代表杠杆做多的力量，融券余额代表做空的力量。我会计算每日的融资净买入额，以及融资余额增速，这两个指标对判断短期市场情绪非常有效。

3. 社交媒体情绪
这是最有意思的部分。我用akshare抓取东方财富股吧的发帖数据，通过关键词匹配（涨、跌、牛、熊、亏、割肉等），计算一个日度情绪指数。这个指数在极端值区间，往往是反向指标。

数据采集是整个系统的基础。我用Python的asyncio构建了一个异步并发采集框架，可以在秒级时间内同时抓取多个数据源，相比同步方式效率提升10倍以上。

下面是这个采集模块的核心代码结构：

import asyncio
import aiohttp
import akshare as ak
import pandas as pd
from datetime import datetime, timedelta
import json
from typing import List, Dict, Optional

class DataCollector:
    \"\"\"异步并发数据采集器\"\"\"
    
    def __init__(self):
        self.session: Optional[aiohttp.ClientSession] = None
        self.cache: Dict[str, pd.DataFrame] = {}
    
    async def __aenter__(self):
        self.session = aiohttp.ClientSession(
            headers={"User-Agent": "Mozilla/5.0"}
        )
        return self
    
    async def __aexit__(self, *args):
        if self.session:
            await self.session.close()
    
    async def fetch_north_money(self) -> pd.DataFrame:
        \"\"\"北向资金流向数据\"\"\"
        loop = asyncio.get_event_loop()
        # akshare的同步函数需要在线程里跑
        df = await loop.run_in_executor(
            None, 
            ak.stock_em_hsgt_north_net_flow_in
        )
        return df
    
    async def fetch_margin_balance(self) -> pd.DataFrame:
        \"\"\"融资融券余额\"\"\"
        loop = asyncio.get_event_loop()
        df = await loop.run_in_executor(
            None,
            ak.stock_margin_detail
        )
        return df
    
    async def collect_all(self) -> Dict[str, pd.DataFrame]:
        \"\"\"并发采集所有数据\"\"\"
        north, margin = await asyncio.gather(
            self.fetch_north_money(),
            self.fetch_margin_balance()
        )
        return {
            "north_money": north,
            "margin": margin
        }

这个设计的核心是用asyncio.gather实现真正的并发采集。akshare是同步函数，所以需要用run_in_executor把同步函数扔到线程池里跑，主线程继续执行其他任务。

在我的实测中，采集北向资金+融资融券+情绪数据全套，在4M宽带下只需要约8秒，而如果用同步串行方式需要约45秒。差距接近6倍。

采集上来的原始数据，需要经过加工才能变成有用的因子。我重点讲三个核心因子的计算逻辑。

因子一：北向资金情绪偏离度

这个因子的设计思路是：每天计算北向资金的净流入额，然后计算过去20日的移动平均值，用当日值减去均值，再除以标准差，得到一个Z-Score。当Z-Score超过2的时候，说明外资当天买入力度异常强，可能是趋势加速信号；当Z-Score低于-2时，说明外资在集中抛售。

import numpy as np

def calc_north_zscore(df: pd.DataFrame, 
                      col: str = "北向资金净流入",
                      window: int = 20) -> pd.Series:
    \"\"\"计算北向资金Z-Score偏离度\"\"\"
    net_flow = df[col].astype(float)
    
    # 计算滚动均值和标准差
    rolling_mean = net_flow.rolling(window=window, min_periods=10).mean()
    rolling_std = net_flow.rolling(window=window, min_periods=10).std()
    
    # Z-Score = (当日值 - 均值) / 标准差
    zscore = (net_flow - rolling_mean) / rolling_std
    
    return zscore

# 使用示例
north_df["zscore"] = calc_north_zscore(north_df)
print(f"4月16日北向资金Z-Score: {north_df['zscore'].iloc[-1]:.2f}")

因子二：融资杠杆情绪指数

融资余额的日变化率，是一个极其灵敏的情绪指标。我的计算方式是：每天计算融资余额的环比增长率（今日/昨日-1），然后对过去5日的增长率做加权平均（近权重更高），得到一个平滑后的融资情绪指数。

def calc_margin_sentiment(df: pd.DataFrame,
                          col: str = "融资余额",
                          ewm_span: int = 5) -> pd.Series:
    \"\"\"计算融资情绪指数（指数加权移动平均）\"\"\"
    balance = df[col].astype(float)
    
    # 日环比增长率
    daily_change = balance.pct_change()
    
    # 指数加权移动平均（近期权重更大）
    sentiment = daily_change.ewm(span=ewm_span).mean()
    
    return sentiment

margin_df["margin_sentiment"] = calc_margin_sentiment(margin_df)
# 数值大于0说明杠杆资金在加仓
# 数值小于0说明杠杆资金在减仓

因子三：股吧情绪指数

这是最复杂的一个因子，也是最有意思的。我用akshare的股吧帖子数据接口，抓取热门讨论帖，然后根据正面/负面关键词的出现频率计算情绪得分。

import re

POSITIVE_WORDS = ["涨停", "暴赚", "大涨", "牛市", "翻倍", 
                  "吃肉", "满仓", "抄底", "稳了"]
NEGATIVE_WORDS = ["跌停", "暴亏", "大跌", "熊市", "腰斩",
                  "割肉", "清仓", "被套", "凉了"]

def calc_forum_sentiment(posts: list) -> float:
    \"\"\"计算论坛情绪得分 [-1, 1]\"\"\"
    if not posts:
        return 0.0
    
    pos_count = 0
    neg_count = 0
    
    for post in posts:
        content = post.lower()
        for w in POSITIVE_WORDS:
            pos_count += content.count(w)
        for w in NEGATIVE_WORDS:
            neg_count += content.count(w)
    
    total = pos_count + neg_count
    if total == 0:
        return 0.0
    
    # 返回 [-1, 1] 区间的情绪得分
    score = (pos_count - neg_count) / total
    return score

因子计算完之后，需要把多个因子综合起来生成一个可读的交易信号。我的综合信号生成逻辑如下：

def generate_signal(north_zscore: float,
                    margin_sentiment: float,
                    forum_sentiment: float,
                    rsi: float,
                    threshold: float = 1.0) -> dict:
    \"\"\"
    综合多因子生成交易信号
    返回: {"signal": "买入/中性/卖出", "confidence": 0.0~1.0}
    \"\"\"
    signals = []
    
    # 北向资金因子
    if north_zscore > threshold:
        signals.append(("north", 1, north_zscore / 3))
    elif north_zscore < -threshold:
        signals.append(("north", -1, abs(north_zscore) / 3))
    
    # 融资情绪因子（连续3天正值更有力）
    if margin_sentiment > 0.001:
        signals.append(("margin", 1, min(margin_sentiment * 100, 1.0)))
    elif margin_sentiment < -0.001:
        signals.append(("margin", -1, min(abs(margin_sentiment) * 100, 1.0)))
    
    # 论坛情绪逆向指标
    if forum_sentiment < -0.3:
        # 情绪极度悲观，可能是反向买入信号
        signals.append(("forum", 1, abs(forum_sentiment)))
    elif forum_sentiment > 0.3:
        signals.append(("forum", -1, forum_sentiment))
    
    # RSI极端值
    if rsi > 75:
        signals.append(("rsi", -1, (rsi - 75) / 25))
    elif rsi < 30:
        signals.append(("rsi", 1, (30 - rsi) / 30))
    
    # 加权综合
    total_weight = sum(s[2] for s in signals)
    if total_weight == 0:
        return {"signal": "中性", "confidence": 0.0, "details": signals}
    
    weighted = sum(s[1] * s[2] for s in signals) / total_weight
    
    if weighted > 0.3:
        final_signal = "偏多"
    elif weighted < -0.3:
        final_signal = "偏空"
    else:
        final_signal = "中性"
    
    return {
        "signal": final_signal,
        "confidence": abs(weighted),
        "raw_score": weighted,
        "details": signals
    }

最后一步是可视化。我用mplfinance把K线图和量价数据画出来，然后在上面叠加我计算出的各个因子信号。这样一眼就能看出，当前市场的量价关系和情绪因子是否共振。

import mplfinance as mpf

def plot_chi_next_analysis(price_df: pd.DataFrame,
                            signals: dict,
                            output_path: str):
    \"\"\"绘制创业板综合分析图\"\"\"
    
    # 添加情绪信号作为副图
    apds = [
        mpf.make_addplot(price_df["north_zscore"], 
                        panel=2, color="blue", 
                        ylabel="北向ZScore"),
        mpf.make_addplot(price_df["margin_sentiment"], 
                        panel=3, color="green",
                        ylabel="融资情绪"),
    ]
    
    mc = mpf.make_marketcolors(
        up='#ff4136', down='#3d9970',
        edge='inherit', wick='inherit'
    )
    style = mpf.make_mpf_style(marketcolors=mc)
    
    mpf.plot(price_df, type='candle', style=style,
             title='创业板指 CHI NEXT - 4月16日',
             ylabel='Price',
             addplot=apds,
             savefig=output_path,
             figsize=(14, 10))
    
    print(f"图表已保存: {output_path}")

回到4月16日。收盘后我运行这套系统，输出了当天的一组数据：

北向资金Z-Score: +2.7
这是一个极强的信号。4月16日外资净流入规模，远超过去20日的均值加两个标准差。这通常意味着有实质性利好催化，而不是简单的情绪波动。

融资情绪指数: +0.0032
杠杆资金在过去5天的日均增速转正，说明做多力量在加仓。但这个信号并不极端，属于正常偏多区间。

综合信号评分: +0.65（偏多）
系统给出的结论是：市场情绪偏多，但北向资金是主要驱动力，融资杠杆的配合力度中等。这种分化意味着：行情的主要推手是外资配置盘，而不是散户杠杆资金。推动力结构相对健康。

这也解释了为什么创业板能在RSI接近80的情况下，依然保持了较强的上行动能——因为推动力来自理性的机构资金，而不是情绪驱动的散户追涨。

# 系统运行日志（4月16日收盘）
数据采集耗时: 8.3秒
北向Z-Score: +2.70 [强烈积极]
融资情绪: +0.0032 [偏多]
综合信号: 偏多(confidence=0.65)
建议: 趋势方向偏多，注意RSI极端值风险

这套系统的代码超过1000行，本文展示的是核心架构。如果你想获取完整代码，关注公众号后留言"监控系统"，我会把完整项目地址发给你。