在股票交易中,筹码分布是判断主力建仓成本和持仓结构的重要工具。但传统分析方法依赖人工经验,需要解读每个价位的成交占比、识别筹码峰的迁移、判断支撑压力位,对新手来说门槛较高。
本教程将带你用AI Agent构建一个筹码分布分析系统,让机器自动从分价成交、历史成交分布、资金流向等数据中还原主力的建仓过程。系统采用模块化设计,每个子任务交由专门的Analyzer处理,最后汇总生成综合判断,既便于优化,也方便扩展新的分析维度。
系统架构与设计思路
系统的核心是将筹码分析拆解成四个独立的子任务,每个子任务由一个专门的Analyzer负责。这种设计的好处是:
- 可以单独优化每个分析器,比如调整参数或算法
- 便于后续扩展新的分析维度,比如加入成交量异动分析
- 判断标准统一,不会因为情绪波动影响决策
数据层使用本地数据引擎,主要用到四个接口:
- 当天分价成交占比接口:time/real/trace/realpercent/{dm},返回每个价格对应的成交量和占比,是计算当日筹码分布的基础
- 历史成交分布接口:time/zijin/lscjfb/{dm},提供近十年每天的成交分布数据,包括超大单、大单、小单、散单的流入流出,用于分析筹码的长期演化
- 最近10天成交分布接口:time/zijin/lscjfb/last10/{dm},是历史成交分布的短期版本,用于捕捉近期的筹码变动
- 实时行情接口:time/real/{dm},提供当前价格、涨跌幅等实时数据,用于判断筹码的当前状态
class ChipDistributionAgent:
def __init__(self, data_engine):
self.data_engine = data_engine
self.analyzers = [
PriceLevelAnalyzer(),
ChipMigrationAnalyzer(),
CapitalFlowAnalyzer(),
SupportResistanceAnalyzer()
]
def analyze(self, dm):
realpercent = self.data_engine.read_realpercent(dm)
lscjfb = self.data_engine.read_lscjfb(dm)
last10 = self.data_engine.read_lscjfb_last10(dm)
realtime = self.data_engine.read_realtime(dm)
context = {
"dm": dm,
"realpercent": realpercent,
"lscjfb": lscjfb,
"last10": last10,
"realtime": realtime
}
results = []
for analyzer in self.analyzers:
result = analyzer.analyze(context)
if result:
results.append(result)
summary = self.summarize(results, context)
return summary
核心分析器详解
1. 价位成交分析器:识别筹码密集区
第一个分析器是价位成交分析器。它处理当天分价成交占比数据,计算每个价位的成交量和占比,从而识别筹码密集区和筹码稀疏区。
筹码密集区意味着大量成交集中在这个价位,往往是主力的成本区或者关键支撑压力位。通过分析这些密集区,可以判断当前股价处于什么位置。
class PriceLevelAnalyzer:
def analyze(self, context):
realpercent = context["realpercent"]
if not realpercent or len(realpercent) == 0:
return None
total_vol = sum(item["成交量(股)"] for item in realpercent)
levels = []
for item in realpercent:
price = item["成交价"]
vol = item["成交量(股)"]
pct = item["占比(%)"]
levels.append({"price": price, "vol": vol, "pct": pct})
levels_sorted = sorted(levels, key=lambda x: x["pct"], reverse=True)
dense_zones = levels_sorted[:5]
current_price = context["realtime"].get("当前价格", 0)
above_current = sum(l["vol"] for l in levels if l["price"] > current_price)
below_current = sum(l["vol"] for l in levels if l["price"] < current_price)
profit_ratio = below_current / total_vol if total_vol > 0 else 0
signal = {
"analyzer": "price_level",
"dense_zones": dense_zones,
"profit_ratio": profit_ratio,
"above_pressure": above_current,
"below_support": below_current
}
if profit_ratio > 0.7:
signal["signal"] = "大部分筹码获利,套牢盘少"
signal["score"] = 70
elif profit_ratio < 0.3:
signal["signal"] = "大部分筹码套牢,上方压力大"
signal["score"] = 40
else:
signal["signal"] = "筹码分布均衡"
signal["score"] = 55
return signal
获利盘比例是一个很有参考价值的指标。当大部分筹码处于获利状态时,说明股价已经脱离主力成本区,继续上涨的阻力较小;当大部分筹码套牢时,上方压力大,每次反弹都会遇到解套盘的抛压。
小提示:获利盘比例要结合股价位置来看,低位高获利盘与高位高获利盘的含义完全不同。
2. 筹码迁移分析器:判断主力真实意图
第二个分析器是筹码迁移分析器。它对比最近10天的成交分布数据,观察筹码是从低位向高位迁移(主力拉升出货),还是从高位向低位迁移(主力打压建仓)。
筹码迁移分析能帮我们看清楚主力的真实意图。同样是上涨,如果是主力加仓推动的,后续大概率还会涨;如果是散户跟风推动的,往往涨一天就回调。通过对比大单和散单的流入流出变化,可以大致判断当前是主力在主导还是散户在主导。
class ChipMigrationAnalyzer:
def analyze(self, context):
last10 = context["last10"]
if not last10 or len(last10) < 5:
return None
sorted_data = sorted(last10, key=lambda x: x["时间yyyy-MM-dd"])
early = sorted_data[:3]
late = sorted_data[-3:]
early_big_inflow = sum(d.get("超大单净流入(元)", 0) + d.get("大单净流入(元)", 0) for d in early)
late_big_inflow = sum(d.get("超大单净流入(元)", 0) + d.get("大单净流入(元)", 0) for d in late)
early_retail_inflow = sum(d.get("散单净流入(元)", 0) + d.get("小单净流入(元)", 0) for d in early)
late_retail_inflow = sum(d.get("散单净流入(元)", 0) + d.get("小单净流入(元)", 0) for d in late)
price_early = sum(d.get("收盘价(元)", 0) for d in early) / len(early)
price_late = sum(d.get("收盘价(元)", 0) for d in late) / len(late)
price_change = (price_late - price_early) / price_early if price_early > 0 else 0
signal = {
"analyzer": "chip_migration",
"big_inflow_early": early_big_inflow,
"big_inflow_late": late_big_inflow,
"price_change": price_change
}
if early_big_inflow > 0 and late_big_inflow < 0 and price_change > 0.05:
signal["signal"] = "主力低位建仓后高位出货"
signal["score"] = 35
signal["direction"] = "bearish"
elif early_big_inflow < 0 and late_big_inflow > 0 and price_change < 0:
signal["signal"] = "主力打压吸筹"
signal["score"] = 75
signal["direction"] = "bullish"
elif late_big_inflow > early_big_inflow and price_change > 0:
signal["signal"] = "主力持续加仓"
signal["score"] = 70
signal["direction"] = "bullish"
else:
signal["signal"] = "筹码迁移不明显"
signal["score"] = 50
signal["direction"] = "neutral"
return signal
小提示:筹码迁移分析的信号比单日资金流向更可靠,因为主力的建仓出货是一个过程,单看一天容易被误导。
3. 资金流向分析器:判断资金性质与强度
第三个分析器是资金流向分析器。它从历史成交分布数据中提取超大单、大单、小单、散单的净流入情况,判断资金的性质和强度。
资金流向分析里有一个关键概念叫大单主导度。当大单成交占比超过60%时,说明这只股票的主力资金活跃度高,后续出现大行情的概率较大;当大单主导度低于30%时,说明主要是散户在交易,行情往往比较平淡。
class CapitalFlowAnalyzer:
def analyze(self, context):
lscjfb = context["lscjfb"]
if not lscjfb or len(lscjfb) == 0:
return None
recent = lscjfb[:20] if len(lscjfb) >= 20 else lscjfb
big_net = sum(d.get("超大单净流入(元)", 0) + d.get("大单净流入(元)", 0) for d in recent)
retail_net = sum(d.get("散单净流入(元)", 0) + d.get("小单净流入(元)", 0) for d in recent)
big_total = sum(abs(d.get("超大单净流入(元)", 0)) + abs(d.get("大单净流入(元)", 0)) for d in recent)
retail_total = sum(abs(d.get("散单净流入(元)", 0)) + abs(d.get("小单净流入(元)", 0)) for d in recent)
big_dominance = big_total / (big_total + retail_total) if (big_total + retail_total) > 0 else 0
signal = {
"analyzer": "capital_flow",
"big_net": big_net,
"retail_net": retail_net,
"big_dominance": big_dominance
}
if big_net > 0 and retail_net < 0:
signal["signal"] = "主力吸筹,散户出逃"
signal["score"] = 75
signal["direction"] = "bullish"
elif big_net < 0 and retail_net > 0:
signal["signal"] = "主力出货,散户接盘"
signal["score"] = 30
signal["direction"] = "bearish"
elif big_net > 0 and retail_net > 0:
signal["signal"] = "资金共振流入"
signal["score"] = 65
signal["direction"] = "bullish"
else:
signal["signal"] = "资金流向不明确"
signal["score"] = 50
signal["direction"] = "neutral"
return signal
小提示:大单主导度这个指标对中大盘股效果较好,对小盘股效果一般,因为小盘股的大单定义阈值相对偏高。
4. 支撑压力分析器:识别关键价位
第四个分析器是支撑压力分析器。它结合价位成交分布和当前价格,识别关键的支撑位和压力位。支撑位是筹码密集且在当前价格下方的价位,压力位是筹码密集且在当前价格上方的价位。
支撑压力位的盈亏比是决定是否入场的重要依据。如果上方空间是下方风险的两倍以上,这笔交易就值得做;反过来,如果上方压力近在眼前而下方支撑很远,就应该回避。
class SupportResistanceAnalyzer:
def analyze(self, context):
realpercent = context["realpercent"]
realtime = context["realtime"]
if not realpercent or not realtime:
return None
current_price = realtime.get("当前价格", 0)
if current_price == 0:
return None
levels = [(item["成交价"], item["成交量(股)"]) for item in realpercent]
levels.sort(key=lambda x: x[1], reverse=True)
support = None
resistance = None
for price, vol in levels:
if price < current_price and support is None:
support = price
if price > current_price and resistance is None:
resistance = price
if support and resistance:
break
signal = {
"analyzer": "support_resistance",
"support": support,
"resistance": resistance,
"current_price": current_price
}
if support and resistance:
risk_reward = (resistance - current_price) / (current_price - support) if (current_price - support) > 0 else 0
signal["risk_reward"] = risk_reward
if risk_reward > 2:
signal["signal"] = "盈亏比 fa vorable,上方空间大"
signal["score"] = 70
elif risk_reward < 0.5:
signal["signal"] = "盈亏比 unfa vorable,上方压力大"
signal["score"] = 35
else:
signal["signal"] = "盈亏比均衡"
signal["score"] = 50
else:
signal["signal"] = "支撑压力位不明确"
signal["score"] = 50
return signal
综合汇总机制:加权判断
Agent最后会把四个分析器的结果汇总成一个综合判断。汇总时不是简单地取平均,而是根据各分析器的置信度和信号方向做加权处理。比如筹码迁移分析的信号如果和资金流向分析一致,置信度会提升;如果两者矛盾,Agent会降低整体置信度并提示需要更多观察。
def summarize(self, results, context):
if not results:
return {"score": 50, "signal": "数据不足,无法判断"}
total_score = 0
total_weight = 0
signals = []
weights = {
"price_level": 0.25,
"chip_migration": 0.35,
"capital_flow": 0.25,
"support_resistance": 0.15
}
for result in results:
analyzer = result.get("analyzer", "")
weight = weights.get(analyzer, 0.1)
score = result.get("score", 50)
total_score += score * weight
total_weight += weight
signals.append(result.get("signal", ""))
final_score = total_score / total_weight if total_weight > 0 else 50
if final_score >= 65:
recommendation = "筹码结构 fa vorable,可关注"
elif final_score <= 40:
recommendation = "筹码结构 unfa vorable,谨慎"
else:
recommendation = "筹码结构中性,继续观察"
return {
"dm": context["dm"],
"score": round(final_score, 1),
"recommendation": recommendation,
"signals": signals,
"current_price": context["realtime"].get("当前价格", 0)
}
实战经验与使用技巧
在实际使用中,我会让Agent每天盘后扫描全市场,对所有股票做一遍筹码分析,筛选出综合得分高于65分的股票作为观察池。然后再结合基本面和技术面做二次筛选。这种Agent驱动的分析方式比人工逐只看筹码图效率高得多,而且判断标准统一,不会因为情绪波动而影响决策。
以下是几个值得分享的经验:
- 筹码迁移分析的信号比单日资金流向更可靠,因为主力的建仓出货是一个过程,单看一天容易被误导
- 获利盘比例要结合股价位置来看,低位高获利盘和高位高获利盘的含义完全不同
- 大单主导度这个指标对中大盘股效果较好,对小盘股效果一般,因为小盘股的大单定义阈值相对偏高
常见问题解答
Q1:数据不足时怎么办?
如果某个分析器返回None,Agent会自动跳过该分析器,只使用其他有效数据的加权结果。如果所有分析器都返回None,则返回默认的50分中性信号。
Q2:信号之间矛盾怎么办?
当筹码迁移分析和资金流向分析的信号矛盾时,Agent会降低整体置信度。建议此时多观察几天,等待信号一致后再做决策。
Q3:这个系统适合哪些市场?
系统适用于A股市场,数据来自ig50本地数据接口,数据更新及时,字段完整,适合构建多维度分析系统。对于其他市场,需要调整数据接口和参数。
Q4:如何优化权重分配?
默认权重分配是:筹码迁移0.35、价位成交0.25、资金流向0.25、支撑压力0.15。你可以根据市场环境调整,比如在震荡市中提高支撑压力的权重,在趋势市中提高筹码迁移的权重。
系统使用的数据来自ig50本地数据接口,数据更新及时,字段完整,适合构建这种多维度分析系统。
