用DeepSeek编写的主力洗盘主升开启的程序

一个基于量价、资金、筹码和技术指标判断主力洗盘及主升开启的Python程序框架。实际应用中需接入实时数据并调整参数：

“`python

import pandas as pd

import talib as ta

import numpy as np

class MainTrendStrategy:

def __init__(self):

self.volatility_threshold = 0.15 # 价格波动率阈值

self.volume_ratio = 0.7 # 洗盘阶段成交量比前期阈值

self.capital_inflow_threshold = 0.5 # 主力资金流入阈值（百分比）

def calculate_indicators(self, df):

# 计算技术指标

df[‘MACD’], df[‘MACDsignal’], _ = ta.MACD(df[‘close’])

df[‘RSI’] = ta.RSI(df[‘close’])

df[‘KDJ_K’], df[‘KDJ_D’] = ta.STOCH(df[‘high’], df[‘low’], df[‘close’])

df[‘MA5’] = ta.SMA(df[‘close’], timeperiod=5)

df[‘MA20’] = ta.SMA(df[‘close’], timeperiod=20)

# 计算波动率（示例逻辑）

df[‘volatility’] = df[‘high’]/df[‘low’] – 1

# 计算量能变化（示例）

df[‘volume_ma5’] = ta.SMA(df[‘volume’], 5)

df[‘volume_change’] = df[‘volume_ma5’]/df[‘volume_ma5’].shift(3)

return df

def detect_washout(self, df, window=10):

“””

主力洗盘阶段识别

条件：

1. 价格震荡但趋势未破

2. 成交量萎缩

3. 筹码集中度提升

4. MACD底背离

5. 主力资金保持流入

“””

df[‘washout_signal’] = 0

for i in range(window, len(df)):

# 价格波动率条件

volatility = df[‘volatility’].iloc[i-window:i].mean()

# 成交量条件

volume_condition = df[‘volume_change’].iloc[i] < self.volume_ratio

# MACD底背离检测（简化版）

macd_divergence = (df[‘close’].iloc[i] < df[‘close’].iloc[i-3]) and \

(df[‘MACD’].iloc[i] > df[‘MACD’].iloc[i-3])

# 资金面条件（假设有主力资金数据）

capital_condition = df[‘main_capital’].iloc[i] > self.capital_inflow_threshold

# 筹码集中度（假设有筹码数据）

chip_concentration = df[‘chip_concentration’].iloc[i] > df[‘chip_concentration’].iloc[i-5]

if volatility > self.volatility_threshold and \

volume_condition and \

macd_divergence and \

capital_condition and \

chip_concentration:

df[‘washout_signal’].iloc[i] = 1

return df

def detect_main_rise(self, df):

“””

主升浪开启识别

条件：

1. 放量突破压力位

2. 均线多头排列

3. MACD金叉

4. 主力资金大幅流入

5. 政策利好

“””

df[‘main_rise_signal’] = 0

for i in range(1, len(df)):

# 量价突破条件

volume_break = df[‘volume’].iloc[i] > 1.5 * df[‘volume_ma5’].iloc[i]

price_break = df[‘close’].iloc[i] > df[‘high’].iloc[i-5:i].max()

# 均线排列

ma_condition = df[‘MA5’].iloc[i] > df[‘MA20’].iloc[i]

# MACD金叉

macd_golden = (df[‘MACD’].iloc[i] > df[‘MACDsignal’].iloc[i]) and \

(df[‘MACD’].iloc[i-1] < df[‘MACDsignal’].iloc[i-1])

# 资金条件

capital_condition = df[‘main_capital’].iloc[i] > 2 * self.capital_inflow_threshold

# 政策条件（示例：行业利好政策）

policy_condition = df[‘policy_news’].iloc[i] == 1 # 假设有政策标记

if volume_break and price_break and ma_condition and \

macd_golden and capital_condition and policy_condition:

df[‘main_rise_signal’].iloc[i] = 1

return df

def run_strategy(self, data):

df = self.calculate_indicators(data)

df = self.detect_washout(df)

df = self.detect_main_rise(df)

return df

# 示例使用

if __name__ == “__main__”:

# 需要包含以下字段的DataFrame（示例数据需自行补充）：

# – OHLC数据

# – volume成交量

# – main_capital主力资金流向

# – chip_concentration筹码集中度

# – policy_news政策利好标记

strategy = MainTrendStrategy()

result_df = strategy.run_strategy(your_input_dataframe)

# 输出信号点

washout_points = result_df[result_df[‘washout_signal’] == 1]

main_rise_points = result_df[result_df[‘main_rise_signal’] == 1]

“`

需要注意的关键点：

1. 数据需求：

– 需要接入主力资金流向数据（大单资金监控）

– 筹码集中度数据（来自定期报告或L2数据）

– 政策新闻数据（需要NLP处理或人工标记）

2. 参数优化建议：

– 波动率阈值应根据不同股票调整

– MACD参数可优化(默认12,26,9)

– 回测时应测试不同市场周期表现

3. 增强方向：

– 加入行业板块轮动分析

– 增加筹码分布形态识别

– 结合订单流数据分析

– 加入风险控制模块

4. 信号验证：

– 主升浪信号需满足洗盘后20个交易日内出现

– 突破需伴随至少3日站稳压力位

– 应排除高位放量情形

实际部署时需要：

1. 接入实时数据源

2. 进行历史回测优化参数

3. 设置止损止盈机制

4. 结合基本面分析过滤标的

建议先用历史数据验证策略有效性，再结合实时行情使用。不同市场环境下可能需要动态调整参数阈值。