量化研究---基于东财的人气100排序自动导入通达信（附源码）

admin · 发表于 2025-3-23 09:03:25

大家好，我是仓鼠哥，专注于指标评测及量化研究！

全自动量化工具，大数据选股，智能风控，AI盯盘，无需人工干预，每个人都能用量化交易！！！

关注仓鼠哥量化研究，围观机器人量化实盘:

量化研究：基于东财人气100排序自动导入通达信（附源码）

在量化交易中，实时追踪市场情绪与资金动向是制定策略的重要依据。东方财富网（EastMoney）的“人气100”榜单反映了当前A股市场关注度最高的个股排名，为我们提供了散户情绪的直观参考。

然而，手动记录并分析这些数据效率低下，因此，本文介绍一种基于Python的自动化解决方案，实现东财人气榜单的实时抓取、清洗，并一键导入通达信软件，助力投资者快速构建量化分析模型。

之前给老铁们分享过网页抓取人气股排名，今天研究了下，可以直接通过PY将人气股导入到通达信自定义板块，不用去充值会员生成了，当然有懂的朋友可以进一步研究衍生成其他的通达信结合方式。一、实现原理

1. 数据抓取：通过Python的requests库模拟浏览器访问东财人气榜单页面（如：http://data.eastmoney.com/stock/lhb.html），利用`BeautifulSoup`或正则表达式解析HTML源码，提取股票代码、名称、人气值等关键字段。
2. 通达信导入：通达信支持通过自定义数据格式（如.txt或特定目录的Excel）导入板块信息。通过脚本将清洗后的数据自动复制到通达信的自定义板块目录下（T0002/blocknew），实现实时同步。

二、应用场景

1. 情绪监控：每日开盘前自动更新人气榜单，识别市场热点。
2. 策略回测：结合历史人气数据与股价波动，验证“追涨”或“反转”策略的有效性。
3. 盘中预警：通过通达信条件选股功能，实时监控榜单个股异动。

三、注意事项

• 合规性：频繁爬取可能导致IP被封禁，建议控制请求频率或使用官方API（如有权限）。
• 数据延迟：东财榜单更新存在分钟级延迟，需结合实时行情数据使用。
• 路径配置：需根据本地通达信安装目录调整文件保存路径。

四、使用方法

1、解压缩下面文件夹后，打开分享配置文件

2、设置配置文件中TDX地址，就可以了。

3、然后在phyhon中运行“通达信自选股添加.py”即自动将东财人气股自动添加到了通达信自定义办卡中了！

本文提供的自动化脚本（附完整源码）打通了东财人气数据与通达信的量化分析链路，用户可在此基础上扩展更多功能（如自动选股、邮件通知等）。通过技术与策略的结合，大家能够更高效地捕捉市场机会，提升决策科学性。

详细源码打包见后获取附件：

通达信效果如下

PY运行效果见下

策略源代码可以自己测试看看

from dfcfrq import popularity
from csg_tdx import tdx
import schedule
import time
import json
import requests
import pandas as pd
import akshare as ak
class tdx_stock:
def __init__(self,path=r'E:/new_tdx/T0002/blocknew',name='CS'):
self.path=path
self.tdx=tdx(self.path)
self.rq=popularity()
self.name=name
def get_rq_data(self):
'''
获取人气数据
'''
df=self.rq.get_all_stock_rank_code()
df['证券代码']=df['证券代码'].apply(lambda x: str(x).split('.')[-1])
print(df)
return df
def get_upper_and_down_amount(self,x1=2,x2=-5):
'''
获取市场上涨和下单数据
大盘风控要求
大盘上涨2%以上家数大于500家
大盘下跌5%以上家数小于500家
当日跌停数量小于30家
（满足此条件才开始交易）
'''
url='https://push2ex.eastmoney.com/getTopicZDFenBu?cb=callbackdata8116174&ut=7eea3edcaed734bea9cbfc24409ed989&dpt=wz.ztzt&_=1700720695419'
res=requests.get(url=url)
text=res.text
text=text[20:len(text)-2]
json_text=json.loads(text)
result=json_text['data']['fenbu']
#print(result)
#跌停
down_amount=result[2]['-11']
#涨停
up_amount=result[-9]['11']
#下跌数量
amount_dowm=0
#上涨数量
amount_up=0
for i in result:
for j in range(x1,12):
j=str(j)
amount_up+=i.get(j,0)
for i in result:
for j in range(-11,x2):
j=str(j)
amount_dowm+=i.get(j,0)
return up_amount,down_amount,amount_up,amount_dowm
def stock_board_concept_name_em(self):
'''
概念板块
'''
df=ak.stock_board_concept_name_em()
return df
def stock_board_concept_cons_em(self,name='昨日涨停'):
'''
-概念板块-板块成份
'''
df=ak.stock_board_concept_cons_em(symbol=name)
return df
def stock_board_industry_name_em(self):
'''
行业板块
'''
df=ak.stock_board_industry_name_em()
return df
def stock_board_industry_cons_em(self,name='教育'):
'''
行业板块-板块成份
'''
df=ak.stock_board_industry_cons_em(symbol=name)
return df
def get_concept_industry_analysis(self,n1=5,n2=10):
'''
行业概念选股分析
'''
with open('分析配置.json','r+',encoding='utf-8') as f:
com=f.read()
text=json.loads(com)
trader_list=text['选择股票开头']
data=pd.DataFrame()
concept=self.stock_board_concept_name_em()[:n1]
print('排名前{}的行业'.format(n1))
print(concept)
for symbol in concept['板块名称']:
df=self.stock_board_concept_cons_em(name=symbol)
df['选择']=df['代码'].apply(lambda x:'是' if str(x)[:2] in trader_list else '不是')
df=df[df['选择']=='是']
data=pd.concat([data,df[:n2]],ignore_index=True)
return data
def get_trader_stock(self):
'''
获取交易的股票池
'''
with open('分析配置.json','r+',encoding='utf-8') as f:
com=f.read()
text=json.loads(com)
n1=text['选择板块前N']
n2=text['板块成分股前N']
n3=text['人气选股前N']
data=self.get_concept_industry_analysis(n1=n1,n2=n2)
print(data)
df=self.get_rq_data()[:n3]
print(df)
stock_list=df['证券代码'].tolist()
for stock in data['代码'].tolist():
stock_list.append(stock)
return stock_list
def add_all_stock(self):
'''
添加全部股票
'''
with open('分析配置.json','r+',encoding='utf-8') as f:
com=f.read()
text=json.loads(com)
up_ratio=text['大盘上涨N%以上']
up_amount_1=text["大盘上涨N%以上家数大于"]
down_ratio=text['大盘下跌N%']
down_amount_1=text["大盘下跌N%以上家数小于"]
amount_dowm_1=text["当日跌停数量小于"]
up_amount,down_amount,amount_up,amount_dowm=self.get_upper_and_down_amount(x1=up_ratio,x2=down_ratio)
#if amount_up>=up_amount_1 and amount_dowm<=down_amount_1 and down_amount<=amount_dowm_1:
aa=1
if aa>0:
print('开启模型涨停数量{} 跌停数量{} 大盘上涨{}%以上家数大于数量{} 大盘下跌{}%以上家数{}'.format(up_amount,down_amount,up_ratio,amount_up,down_ratio,amount_dowm))
stock_list=self.get_trader_stock()
self.tdx.add_tdx_stock_list(name=self.name,user_stock_list=stock_list)
else:
print('不选股')
print('不开启模型涨停数量{} 跌停数量{} 大盘上涨{}%以上家数大于数量{} 大盘下跌{}%以上家数{}'.format(up_amount,down_amount,up_ratio,amount_up,down_ratio,amount_dowm))
if __name__=='__main__':
with open('分析配置.json','r+',encoding='utf-8') as f:
com=f.read()
text=json.loads(com)
path=text['通达信路径']
name=text['操作通达信自选股']
cycle_buy_time=text['循环选股时间']
models=tdx_stock(path=r'{}'.format(path),name=name)
#df=models.add_all_stock()
schedule.every(cycle_buy_time).minutes.do(models.add_all_stock)
while True:
schedule.run_pending()
time.sleep(1)

复制代码

		自动登录	找回密码
密码			立即注册