青龙面板定时:
以下是您需要的Python脚本,它可以在青龙面板中运行,每天收盘后筛选符合条件的股票,并通过企业微信推送结果。脚本逻辑严格按照您描述的通达信指标转换而来。 脚本功能使用前准备
- #!/usr/bin/env python3
- # -*- coding: utf-8 -*-
- """
- 通达信涨停回调再启动策略 - 青龙面板版
- 每天收盘后运行,筛选符合条件的股票并通过企业微信推送
- """
- import pandas as pd
- import numpy as np
- import akshare as ak
- import requests
- import time
- from datetime import datetime, timedelta
- # ==================== 配置参数 ====================
- WECHAT_ROBOT_KEY = "YOUR_WECHAT_ROBOT_KEY" # 替换为你的企业微信机器人key
- LOOKBACK_DAYS = 60 # 获取多少天历史数据
- MAX_ZT_DAYS_AGO = 20 # 涨停日距离今天不超过20天
- MIN_CONTINUOUS_DAYS = 1 # 最少连续在MA5下方天数
- PLATE_LIMIT = { # 不同板块涨停阈值(可根据需要调整)
- '主板': 9.5,
- '创业板': 19.5,
- '科创板': 19.5,
- '北交所': 29.5
- }
- # =================================================
- def get_stock_list():
- """获取A股列表,返回DataFrame包含代码和名称"""
- stock_info = ak.stock_info_a_code_name()
- # 可选:过滤北交所(股票代码以8开头或4开头?这里简单过滤8和4)
- # stock_info = stock_info[~stock_info['code'].str.startswith(('8', '4'))]
- return stock_info
- def get_stock_data(code, end_date=None):
- """
- 获取个股历史行情
- :param code: 股票代码(如 '000001')
- :param end_date: 截止日期,默认今天
- :return: DataFrame包含日期、开盘、收盘、最高、最低、成交量
- """
- if end_date is None:
- end_date = datetime.now().strftime('%Y%m%d')
- start_date = (datetime.now() - timedelta(days=LOOKBACK_DAYS*2)).strftime('%Y%m%d') # 多取一些以确保均线计算
- try:
- df = ak.stock_zh_a_hist(symbol=code, period='daily', start_date=start_date, end_date=end_date, adjust='')
- if df.empty:
- return None
- # 重命名列
- df.rename(columns={
- '日期': 'date',
- '开盘': 'open',
- '收盘': 'close',
- '最高': 'high',
- '最低': 'low',
- '成交量': 'volume'
- }, inplace=True)
- df['date'] = pd.to_datetime(df['date'])
- df.sort_values('date', inplace=True)
- df.reset_index(drop=True, inplace=True)
- return df
- except Exception as e:
- print(f"获取{code}数据失败: {e}")
- return None
- def is_limit_up(row, threshold=9.5):
- """判断是否为涨停(涨幅>threshold且收盘价=最高价)"""
- return row['pct_change'] > threshold and row['close'] == row['high']
- def check_stock(df):
- """
- 核心条件检查
- :param df: 包含足够历史数据的DataFrame,按日期升序
- :return: (是否满足, 涨停日, 涨停最低价, 今日收盘价, 连续下方天数)
- """
- if df is None or len(df) < 20: # 数据太少无法判断
- return False, None, None, None, None
- # 计算涨跌幅和5日均线
- df['pct_change'] = (df['close'] - df['close'].shift(1)) / df['close'].shift(1) * 100
- df['ma5'] = df['close'].rolling(window=5).mean()
- # 识别涨停日(最近MAX_ZT_DAYS_AGO天内)
- recent = df.iloc[-MAX_ZT_DAYS_AGO-5:] # 扩大一点范围
- zt_days = recent[recent.apply(lambda row: is_limit_up(row), axis=1)]
- if zt_days.empty:
- return False, None, None, None, None
- # 取最近的涨停日
- zt_day = zt_days.iloc[-1]
- zt_idx = zt_day.name
- zt_low = zt_day['low'] # 涨停当日最低价
- zt_date = zt_day['date']
- # 涨停日之后的数据
- after_zt = df.loc[zt_idx+1:].copy()
- if after_zt.empty:
- return False, None, None, None, None
- # 查找涨停后首次跌破MA5的日期
- below_ma5_start = after_zt[after_zt['close'] < after_zt['ma5']]
- if below_ma5_start.empty:
- return False, None, None, None, None
- first_below_idx = below_ma5_start.index[0] # 首次跌破MA5的索引
- # 从首次跌破到昨天为止,是否一直未站上MA5且未跌破涨停最低价
- check_period = df.loc[first_below_idx:-2] # 截止到昨天
- if len(check_period) == 0:
- return False, None, None, None, None
- # 条件1: 这段期间所有收盘价都低于MA5(即从未站上)
- all_below_ma5 = (check_period['close'] < check_period['ma5']).all()
- # 条件2: 这段期间的最低价从未跌破涨停最低价
- low_not_break = (check_period['low'].min() > zt_low)
- if not (all_below_ma5 and low_not_break):
- return False, None, None, None, None
- # 连续天数
- continuous_days = len(check_period)
- # 检查今天是否站上MA5(收盘价上穿)
- today = df.iloc[-1]
- yesterday = df.iloc[-2]
- cross_up = (today['close'] > today['ma5']) and (yesterday['close'] <= yesterday['ma5'])
- if cross_up and continuous_days >= MIN_CONTINUOUS_DAYS:
- return True, zt_date.strftime('%Y-%m-%d'), zt_low, today['close'], continuous_days
- else:
- return False, None, None, None, None
- def send_wechat_msg(content):
- """通过企业微信机器人发送消息"""
- url = f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={WECHAT_ROBOT_KEY}"
- headers = {"Content-Type": "application/json"}
- data = {
- "msgtype": "text",
- "text": {
- "content": content
- }
- }
- try:
- response = requests.post(url, json=data, headers=headers)
- if response.status_code == 200:
- print("企业微信推送成功")
- else:
- print(f"企业微信推送失败: {response.text}")
- except Exception as e:
- print(f"企业微信推送异常: {e}")
- def main():
- print(f"开始执行策略筛选,时间:{datetime.now()}")
- stock_list = get_stock_list()
- total = len(stock_list)
- print(f"获取到 {total} 只股票")
- matched_stocks = []
- for idx, row in stock_list.iterrows():
- code = row['code']
- name = row['name']
- # 补齐6位代码(akshare可能返回不带前导零的代码?)
- code = code.zfill(6)
- print(f"正在处理 [{idx+1}/{total}] {code} - {name}")
- df = get_stock_data(code)
- if df is None:
- continue
- # 检查条件
- passed, zt_date, zt_low, today_close, days = check_stock(df)
- if passed:
- matched_stocks.append({
- '代码': code,
- '名称': name,
- '涨停日': zt_date,
- '涨停最低价': zt_low,
- '今日收盘': today_close,
- '连续下方天数': days
- })
- # 礼貌停顿,避免请求过快
- time.sleep(0.2)
- # 推送结果
- if matched_stocks:
- msg = f"【涨停回调再启动】符合条件的股票 ({len(matched_stocks)}只)\n"
- for s in matched_stocks:
- msg += f"\n{s['代码']} {s['名称']} 涨停日:{s['涨停日']} 回调{days}天 现价:{s['今日收盘']:.2f}"
- else:
- msg = "【涨停回调再启动】今日无符合条件的股票"
- send_wechat_msg(msg)
- print(msg)
- if __name__ == "__main__":
- main()
复制代码 注意事项数据源:使用akshare库,首次运行前需在青龙面板安装:pip3 install akshare pandas requests。 涨停阈值:脚本默认使用9.5%作为主板涨停阈值。如果您想区分板块,可以在get_stock_data后根据代码前缀动态设置阈值(如创业板30****使用19.5)。代码中预留了PLATE_LIMIT字典,您可以自行扩展。 运行时间:建议在每天15:30以后执行,此时当日数据已更新完毕。 企业微信机器人:需要先在企微群中添加机器人,获取webhook地址。 性能优化:若股票数量多导致运行时间过长,可考虑使用多线程,但需注意akshare的访问限制。当前单线程加0.2秒延时,约需15-20分钟,青龙面板通常允许。
如果遇到任何问题(如数据获取失败),请检查网络或更换数据源(如tushare)。您也可以根据实际需求调整参数(如回调天数、涨停范围等)。
|