date: 2025-10-13 updated: 2025-10-13 conversation_id: b204333c-0c5e-4e64-a797-088990020c79 title: "证券市场复盘程序优化与完善" tags: [deepseek, conversation] ---
证券市场复盘程序优化与完善
创建时间: 2025-10-13 19:44
👤 用户:
import method_all import datetime import time import pandas as pd
def getbasedf(now,tradetimemin,tradetimemax): tradedatedf = methodall.sys.getnjjtradedate(now[0:10],6) tradedate = tradedatedf['tradedate'].max() #获取板块异动时的相应数据 sql = f"""SELECT kplbkdetail.code, kplbkdetail.name, kplbkdetail.bkname, esmovewatch.movetype, esmovewatch.tradetime, njjcattlelist.topnum, njjcattlelist.topscore6, njjcattlelist.type, njjcattlelist.risemax, njjcattlelist.fallmax, njjcattlelist.overhigh, njjcattlelist.highdate, njjcattlelist.lowdate, njjrisewatching.lasttop, njjrisewatching.rise5, njjrisewatching.monthincrease, njjrisewatching.goback, njjrisewatching.riseThree, njjrisewatching.volRise, njjrisewatching.riseUp, njjrisewatching.rise, njjrisewatching.ped, njjrisewatching.istop, njjrisewatching.swing, njjrisewatching.volumeratio, njjrisewatching.turnoverrate, njjrisewatching.mainforce, njjrisewatching.amount FROM kplbkdetail INNER JOIN njjcattlelist ON njjcattlelist.code = kplbkdetail.code and njjcattlelist.tradedate = '{tradedate}' INNER JOIN njjrisewatching ON njjrisewatching.code = kplbkdetail.code and njjrisewatching.tradedate = '{tradedate}' INNER JOIN esmovewatch ON esmovewatch.code = kplbkdetail.code and esmovewatch.tradedate = '{tradedate}' and esmovewatch.tradetime <= '{tradetimemax}' and esmovewatch.tradetime >= '{tradetimemin}' where kplbkdetail.bknum <= 200 and njjcattlelist.type = '半年内' and esmovewatch.movetype in ('大笔买入','火箭发射','快速反弹','封涨停板') order by esmovewatch.tradetime asc """ df = methodall.mysql.sql(sql) #板块热度 #df0 = df[df['tradetime'] <= '09:35:00'] bklist = list(df['bkname'].unique()) bkdf = pd.DataFrame() for bkname in bklist: bkdfsingle = df[df['bkname']==bkname] topnumall = len(bkdfsingle[bkdfsingle['movetype'] == '封涨停板']['code'].unique()) endnumsingle = len(bkdfsingle[bkdfsingle['movetype'] == '封跌停板']['code'].unique()) topnumsingle = len(bkdfsingle[bkdfsingle['movetype'] == '封涨停板']) endnumsingle = len(bkdfsingle[bkdfsingle['movetype'] == '封跌停板']) risenum = len(bkdfsingle[bkdfsingle['movetype'] == '火箭发射']) fallnum = len(bkdfsingle[bkdfsingle['movetype'] == '加速下跌']) + len(bkdfsingle[bkdfsingle['movetype'] == '高台跳水']) backnum = len(bkdfsingle[bkdfsingle['movetype'] == '快速反弹']) buynum = len(bkdfsingle[bkdfsingle['movetype'] == '大笔买入']) sellnum = len(bkdfsingle[bkdfsingle['movetype'] == '大笔卖出']) totalnum = len(bkdfsingle) score = 6topnumsingle + 3risenum + buynum + backnum - sellnum - fallnum - 2*endnumsingle savebk ={ 'bkname': bkname, 'tradedate': tradedate, 'score': score, 'topnum': topnumall, 'risenum': risenum + backnum, 'buynum': buynum, 'totalnum': totalnum, 'sellnum': sellnum, 'fallnum': fallnum, } bkdf = pd.concat([bkdf,pd.DataFrame([savebk],index=[0])]) return([bkdf,df])
def write(): #根据统计周期获取日期 now = str(datetime.datetime.now()) tradedatedf = methodall.sys.getnjjtradedate(now[0:10],11) tradedate = tradedatedf['tradedate'].max() tradedatefirst = tradedatedf['tradedate'].min() sql = f"""SELECT njjcattlelist.topnum, njjcattlelist.topscore6, njjcattlelist.type, njjcattlelist.risemax, njjcattlelist.fallmax, njjcattlelist.overhigh, njjcattlelist.highdate, njjcattlelist.lowdate, njjrisewatching.lasttop, njjrisewatching.rise5, njjrisewatching.monthincrease, njjrisewatching.goback, njjrisewatching.riseThree, njjrisewatching.volRise, njjrisewatching.riseUp, njjrisewatching.rise, njjrisewatching.ped, njjrisewatching.istop, njjrisewatching.swing, njjrisewatching.volumeratio, njjrisewatching.turnoverrate, njjrisewatching.mainforce, njjrisewatching.amount FROM njjcattlelist INNER JOIN njjrisewatching ON njjrisewatching.code = njjcattlelist.code and njjrisewatching.tradedate = njjcattlelist.tradedate where njjcattlelist.tradedate = '{tradedate}' order by njjcattlelist.topscore6 desc """ #获取当前最新的股票列表 dfnow = method_all.mysql.sql(sql)
#获取基础计算数据 basedf = getbasedf(tradedate,'09:28:00','15:05:00') dfbk = basedf[0] dfbk = dfbk.sortvalues('score', ascending=False) dfcode = base_df[1]
#获取近10日的板块统计数据 dfbkhistory = methodall.mysql.sql(f"select * from njjbkhot where tradedate < '{tradedate}' and tradedate >= '{tradedatefirst}' and rank_num <= 10")
#依次分析前五个板块 for i in range(5): bkname = dfbk.iloc[i]['bkname'] #板块近10日历史 bknamehistory = dfbkhistory[dfbkhistory['bkname'] == bkname] if len(bknamehistory) == 0: #说明该板块是新热点 savebk ={ 'bkname': bkname, 'tradedate': tradedate, 'score': dfbk.iloc[i]['score'], 'topnum': dfbk.iloc[i]['topnum'], 'risenum': dfbk.iloc[i]['risenum'], 'buynum': dfbk.iloc[i]['buynum'], } else: #统计该板块近十日的平均得分、累计涨停次数、累计大幅拉升次数 bkscoremean = bknamehistory['score'].mean() bktopnum = bknamehistory['topnum'].sum() bkrisenum = bknamehistory['risenum'].sum()
if len(bknamehistory) > 7: #描述说明该板块是近十日的持续性热点 text = f"{bkname} 持续热点" #找出板块内得分最高的十只股票 bkcode = dfcode[dfcode['bkname'] == bkname] bkcode = bkcode.sortvalues('topscore6', ascending=False) bkcodetop10 = bkcode.head(10) codetop10list = list(bkcodetop10['code'].unique()) #遍历前十只股票，描述当前状态 for code in codetop10list: risenow = dfnow[dfnow['code'] == code]['rise'].values[0] coderow = bkcodetop10[bkcodetop10['code'] == code] name = coderow['name'].values[0] if len(coderow[coderow['movetype'] == '封跌停板']) > 0: #描述该股票触及跌停板 text = f"{code} {name} 触及跌停板" if len(coderow[coderow['movetype'] == '封涨停板']) > 0: #描述该股票触及涨停板 text = f"{code} {name} 触及涨停板"
#找出板块内有涨停状态的股票 bkcodetop = bkcode[bkcode['istop'] == 1] codetoplist = list(bkcodetop['code'].unique()) #得出涨停股票数量 bktopnumnow = len(codetoplist) #对比得出板块内新晋涨停热门股 for code in codetoplist: if code not in codetop10list: risenow = dfnow[dfnow['code'] == code]['rise'].values[0] #说明该股票是该板块内新涨停热门股 name = bkcodetop[bkcodetop['code'] == code]['name'].values[0] #找出板块内近期涨幅较小，有大幅拉升状态的股票 bkcoderiselow = bkcode[bkcode['monthincrease'] <= 40] bkcoderiselow = bkcoderiselow[bkcoderiselow['rise5'] <= 15] bkcoderiselow = bkcoderiselow[bkcoderiselow['movetype'] == '火箭发射'] if len(bkcoderiselow) > 0: #描述板块内有哪些异动股票近期涨幅较小，有大幅拉升。 for code in list(bkcoderiselow['code'].unique()): risenow = dfnow[dfnow['code'] == code]['rise'].values[0] name = bkcoderiselow[bkcoderise_low['code'] == code]['name'].values[0]
#分析之前热门当日不在热门的板块 bktoplist = list(dfbk.head[5]['bkname'].unique()) bkhislist = list(dfbkhistory['bkname'].unique()) for bkname in bkhislist: if bkname not in bktoplist: bknamehistory = dfbkhistory[dfbkhistory['bkname'] == bkname] if len(bknamehistory) > 7: #说明该板块是近十日的持续性热点,但是今天不是热点 savebk ={ 'bkname': bkname, 'tradedate': tradedate, 'score': bknamehistory['score'].mean(), 'topnum': bknamehistory['topnum'].sum(), 'risenum': bknamehistory['risenum'].sum(), 'buynum': bknamehistory['buynum'].sum(), } #依次分析当前板块内得分前十的股票，描述下跌的与依然坚挺上涨的股票 codetop10list = list(bknamehistory.head(20)['code'].unique()) for code in codetop10list: coderow = bknamehistory[bkname_history['code'] == code]
res = write()
你是一个证券产品经理，正在写一个自动复盘的写稿程序，请根据以上代码的结构，帮我丰富内容，完善代码。
💭 思考过程
我们正在编写一个自动复盘的写稿程序。根据代码，我们主要完成以下功能：
- 获取基础数据：包括板块异动数据、股票的基本面和技术面数据。
- 计算板块热度：根据板块内股票的异动类型（如涨停、火箭发射等）计算板块得分。
- 分析热门板块：对前5个热门板块进行详细分析，包括板块内股票的表现。
- 分析之前热门但当前不在热门的板块。
但是，现有代码在最后部分（分析热门板块和之前热门板块）并没有完整实现，尤其是没有生成具体的复盘文字稿。 我们的目标是生成一个复盘文字稿，描述当前市场的热点板块、板块内的个股表现，以及之前热门板块的冷却情况。
我们将按照以下步骤完善代码：
- 在获取基础数据后，对当前热门的板块（前5名）进行详细分析，包括：
- 板块近10日的历史表现（判断是否持续热点）
- 板块内得分前10的股票，并描述其状态（如涨停、跌停等）
- 板块内新晋涨停热门股（不在前10但涨停的股票）
- 板块内近期涨幅较小但出现大幅拉升的股票
- 对之前热门但当前不在前5的板块进行分析，特别是近10日内持续热门的板块，描述其冷却情况，并分析板块内代表性股票的表现。
- 将分析结果以文字稿的形式输出。
由于原代码中很多分析只是写了注释，并没有实际生成文字，我们将实际生成文字描述，并存储到一个字符串中，最后返回这个字符串。
注意：原代码中有些变量可能未定义，我们需要确保代码的完整性。
我们将重构write函数，使其返回一个字符串，包含复盘内容。
由于代码中涉及多个数据查询和处理，我们需要注意可能的数据缺失情况。
让我们开始完善代码。
我来帮你完善这个自动复盘写稿程序。作为证券产品经理，我会从投资分析的角度来丰富内容和优化代码结构。
import method_all
import datetime
import time
import pandas as pd
from typing import List, Tuple, Dict, Any
class MarketRecapWriter:
"""市场复盘自动写稿程序"""
def __init__(self):
self.today = str(datetime.datetime.now())[0:10]
self.analysis_result = []
def get_base_df(self, now: str, trade_time_min: str, trade_time_max: str) -> Tuple[pd.DataFrame, pd.DataFrame]:
"""获取基础数据"""
trade_date_df = method_all.sys.get_njj_trade_date(now[0:10], 6)
trade_date = trade_date_df['trade_date'].max()
# 获取板块异动时的相应数据
sql = f"""SELECT
kpl_bk_detail.code,
kpl_bk_detail.name,
kpl_bk_detail.bk_name,
es_move_watch.move_type,
es_move_watch.trade_time,
njj_cattle_list.top_num,
njj_cattle_list.top_score6,
njj_cattle_list.type,
njj_cattle_list.rise_max,
njj_cattle_list.fall_max,
njj_cattle_list.over_high,
njj_cattle_list.high_date,
njj_cattle_list.low_date,
njj_rise_watching.last_top,
njj_rise_watching.rise5,
njj_rise_watching.month_increase,
njj_rise_watching.go_back,
njj_rise_watching.riseThree,
njj_rise_watching.volRise,
njj_rise_watching.riseUp,
njj_rise_watching.rise,
njj_rise_watching.pe_d,
njj_rise_watching.is_top,
njj_rise_watching.swing,
njj_rise_watching.volume_ratio,
njj_rise_watching.turnover_rate,
njj_rise_watching.main_force,
njj_rise_watching.amount
FROM kpl_bk_detail
INNER JOIN njj_cattle_list ON njj_cattle_list.code = kpl_bk_detail.code
and njj_cattle_list.trade_date = '{trade_date}'
INNER JOIN njj_rise_watching ON njj_rise_watching.code = kpl_bk_detail.code
and njj_rise_watching.trade_date = '{trade_date}'
INNER JOIN es_move_watch ON es_move_watch.code = kpl_bk_detail.code
and es_move_watch.trade_date = '{trade_date}'
and es_move_watch.trade_time <= '{trade_time_max}'
and es_move_watch.trade_time >= '{trade_time_min}'
where kpl_bk_detail.bk_num <= 200
and njj_cattle_list.type = '半年内'
and es_move_watch.move_type in ('大笔买入','火箭发射','快速反弹','封涨停板')
order by es_move_watch.trade_time asc
"""
df = method_all.mysql.sql(sql)
# 计算板块热度
bk_list = list(df['bk_name'].unique())
bk_df = pd.DataFrame()
for bk_name in bk_list:
bk_df_single = df[df['bk_name'] == bk_name]
# 统计各类异动数量
top_num_all = len(bk_df_single[bk_df_single['move_type'] == '封涨停板']['code'].unique())
top_num_single = len(bk_df_single[bk_df_single['move_type'] == '封涨停板'])
end_num_single = len(bk_df_single[bk_df_single['move_type'] == '封跌停板'])
rise_num = len(bk_df_single[bk_df_single['move_type'] == '火箭发射'])
fall_num = len(bk_df_single[bk_df_single['move_type'] == '加速跳水']) + len(bk_df_single[bk_df_single['move_type'] == '高台跳水'])
back_num = len(bk_df_single[bk_df_single['move_type'] == '快速反弹'])
buy_num = len(bk_df_single[bk_df_single['move_type'] == '大笔买入'])
sell_num = len(bk_df_single[bk_df_single['move_type'] == '大笔卖出'])
total_num = len(bk_df_single)
# 计算板块热度得分
score = 6 * top_num_single + 3 * rise_num + buy_num + back_num - sell_num - fall_num - 2 * end_num_single
save_bk = {
'bk_name': bk_name,
'trade_date': trade_date,
'score': score,
'top_num': top_num_all,
'rise_num': rise_num + back_num,
'buy_num': buy_num,
'total_num': total_num,
'sell_num': sell_num,
'fall_num': fall_num,
'stock_count': len(bk_df_single['code'].unique())
}
bk_df = pd.concat([bk_df, pd.DataFrame([save_bk], index=[0])])
return bk_df, df
def analyze_hot_sectors(self, df_bk: pd.DataFrame, df_code: pd.DataFrame, df_now: pd.DataFrame,
df_bk_history: pd.DataFrame) -> List[str]:
"""分析热门板块"""
analysis = []
df_bk_sorted = df_bk.sort_values('score', ascending=False)
analysis.append("## 一、热门板块分析")
analysis.append(f"今日共 {len(df_bk)} 个板块出现异动，前五大热门板块如下：")
for i in range(min(5, len(df_bk_sorted))):
bk_name = df_bk_sorted.iloc[i]['bk_name']
score = df_bk_sorted.iloc[i]['score']
top_num = df_bk_sorted.iloc[i]['top_num']
rise_num = df_bk_sorted.iloc[i]['rise_num']
analysis.append(f"\n### {i+1}. {bk_name}")
analysis.append(f"- 热度得分: {score} | 涨停个股: {top_num} | 拉升个股: {rise_num}")
# 板块历史表现分析
bk_history = df_bk_history[df_bk_history['bk_name'] == bk_name]
if len(bk_history) == 0:
analysis.append("- 📈 **新晋热点**：该板块首次进入热门榜单")
else:
avg_score = bk_history['score'].mean()
trend = "强势上升" if score > avg_score * 1.2 else "平稳" if score > avg_score * 0.8 else "有所回落"
analysis.append(f"- 📊 **历史表现**：近10日平均得分{avg_score:.1f}，当前热度{trend}")
if len(bk_history) >= 7:
analysis.append("- 🔥 **持续热点**：该板块已连续多日保持活跃")
# 板块内个股分析
bk_stocks = df_code[df_code['bk_name'] == bk_name]
self._analyze_sector_stocks(bk_stocks, df_now, analysis)
return analysis
def _analyze_sector_stocks(self, bk_stocks: pd.DataFrame, df_now: pd.DataFrame, analysis: List[str]):
"""分析板块内个股表现"""
# 按评分排序
bk_stocks_sorted = bk_stocks.sort_values('top_score6', ascending=False)
top_10_stocks = bk_stocks_sorted.head(10)
# 涨停股分析
limit_up_stocks = bk_stocks[bk_stocks['is_top'] == 1]
if len(limit_up_stocks) > 0:
analysis.append(f"- 🚀 **涨停个股**：共{len(limit_up_stocks)}只")
for _, stock in limit_up_stocks.head(3).iterrows():
rise = df_now[df_now['code'] == stock['code']]['rise'].values[0]
analysis.append(f" {stock['name']}({stock['code']}) +{rise:.1f}%")
# 龙头股分析
if len(top_10_stocks) > 0:
analysis.append("- 💎 **板块龙头**：")
for i, (_, stock) in enumerate(top_10_stocks.head(3).iterrows()):
rise = df_now[df_now['code'] == stock['code']]['rise'].values[0]
analysis.append(f" {stock['name']}({stock['code']}) 评分{stock['top_score6']} +{rise:.1f}%")
# 低位异动股分析
low_position_stocks = bk_stocks[
(bk_stocks['month_increase'] <= 40) &
(bk_stocks['rise5'] <= 15) &
(bk_stocks['move_type'] == '火箭发射')
]
if len(low_position_stocks) > 0:
analysis.append("- 📈 **低位异动**：以下个股近期涨幅较小但出现大幅拉升")
for _, stock in low_position_stocks.head(2).iterrows():
analysis.append(f" {stock['name']}({stock['code']}) 5日涨幅{stock['rise5']:.1f}%")
def analyze_cold_sectors(self, current_hot_sectors: List[str], df_bk_history: pd.DataFrame) -> List[str]:
"""分析冷却板块"""
analysis = []
analysis.append("\n## 二、板块轮动分析")
# 获取历史热门但今日不在前5的板块
historical_hot = df_bk_history['bk_name'].unique()
cold_sectors = [sector for sector in historical_hot if sector not in current_hot_sectors[:5]]
if not cold_sectors:
analysis.append("今日热点板块与近期热点高度重合，轮动效应不明显。")
return analysis
analysis.append("以下近期热门板块今日热度有所回落：")
for sector in cold_sectors[:3]: # 分析前3个冷却板块
sector_history = df_bk_history[df_bk_history['bk_name'] == sector]
if len(sector_history) >= 5:
avg_score = sector_history['score'].mean()
max_score = sector_history['score'].max()
analysis.append(f"\n### {sector}")
analysis.append(f"- 近10日平均热度: {avg_score:.1f}")
analysis.append(f"- 历史最高热度: {max_score:.1f}")
if len(sector_history) > 7:
analysis.append("- ⚠️ **持续热点降温**：该板块曾连续多日活跃，今日出现调整")
else:
analysis.append("- 🔄 **短期热点轮动**：该板块短期炒作后热度回落")
return analysis
def generate_market_summary(self, df_bk: pd.DataFrame) -> List[str]:
"""生成市场概况"""
summary = []
summary.append("# 市场复盘报告")
summary.append(f"## 交易日期：{self.today}\n")
total_sectors = len(df_bk)
avg_score = df_bk['score'].mean()
total_limit_up = df_bk['top_num'].sum()
summary.append("## 市场概况")
summary.append(f"- 📊 **活跃板块**：共{total_sectors}个板块出现异动")
summary.append(f"- 📈 **平均热度**：{avg_score:.1f}")
summary.append(f"- 🚀 **涨停个股**：全市场共{total_limit_up}只个股涨停")
# 市场情绪判断
if avg_score > 50:
summary.append("- 😊 **市场情绪**：积极活跃")
elif avg_score > 20:
summary.append("- 😐 **市场情绪**：温和整理")
else:
summary.append("- 😟 **市场情绪**：相对冷淡")
return summary
def write_recap(self) -> str:
"""生成完整复盘报告"""
try:
# 获取基础数据
trade_date_df = method_all.sys.get_njj_trade_date(self.today, 11)
trade_date = trade_date_df['trade_date'].max()
trade_date_first = trade_date_df['trade_date'].min()
# 获取当前股票数据
sql = f"""SELECT code, rise FROM njj_rise_watching
WHERE trade_date = '{trade_date}'"""
df_now = method_all.mysql.sql(sql)
# 获取板块数据
base_df = self.get_base_df(self.today, '09:28:00', '15:05:00')
df_bk, df_code = base_df
# 获取历史数据
df_bk_history = method_all.mysql.sql(
f"SELECT * FROM njj_bk_hot WHERE trade_date < '{trade_date}' "
f"AND trade_date >= '{trade_date_first}' AND rank_num <= 10"
)
# 生成报告各部分
report_parts = []
report_parts.extend(self.generate_market_summary(df_bk))
report_parts.extend(self.analyze_hot_sectors(df_bk, df_code, df_now, df_bk_history))
report_parts.extend(self.analyze_cold_sectors(
list(df_bk.sort_values('score', ascending=False)['bk_name']),
df_bk_history
))
# 添加投资建议
report_parts.append("\n## 三、投资建议")
report_parts.append("1. 🎯 **关注主线**：紧跟当前热门板块的龙头个股")
report_parts.append("2. ⚡ **把握轮动**：注意热点板块的轮动节奏")
report_parts.append("3. 📉 **控制风险**：避免追高短期涨幅过大的品种")
report_parts.append("4. 🔍 **深度挖掘**：关注低位启动且有资金关注的个股")
report_parts.append(f"\n---\n*报告生成时间：{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}*")
return "\n".join(report_parts)
except Exception as e:
return f"生成复盘报告时出错：{str(e)}"
def main():
"""主函数"""
writer = MarketRecapWriter()
report = writer.write_recap()
# 保存报告到文件
filename = f"market_recap_{writer.today}.md"
with open(filename, 'w', encoding='utf-8') as f:
f.write(report)
print(f"复盘报告已生成：{filename}")
print("\n" + "="*50)
print(report)
return report
if __name__ == "__main__":
result = main()主要改进内容：
1. 代码结构优化
- 使用面向对象编程，封装成类
- 添加类型注解，提高代码可读性
- 模块化功能，便于维护和扩展
2. 分析内容丰富
- 市场概况：整体活跃度、情绪判断
- 热门板块分析：新晋热点、持续热点、龙头个股
- 板块轮动分析：冷却板块、轮动节奏
- 投资建议：实用的操作策略
3. 投资逻辑完善
- 热度评分体系
- 历史对比分析
- 个股异动识别
- 风险提示
4. 输出格式优化
- 使用Markdown格式，便于阅读
- 添加表情符号，提升可读性
- 结构化输出，层次清晰
5. 异常处理
- 添加try-catch块
- 数据边界检查
- 错误信息提示
这个版本的代码更加完整，能够生成专业的市场复盘报告，适合投资分析使用。
👤 用户:
源代码修改为以下了，帮我再更新以下 import method_all import datetime import time import pandas as pd
def getbasedf(now): tradedatedf = methodall.sys.getnjjtradedate(now[0:10],6) tradedate = tradedatedf['tradedate'].max() #获取板块异动时的相应数据 sql = f"""SELECT kplbkdetail.code, kplbkdetail.name, kplbkdetail.bkname, esmovewatch.movetype, esmovewatch.tradetime, njjrisewatching.istop FROM kplbkdetail INNER JOIN njjrisewatching ON njjrisewatching.code = kplbkdetail.code and njjrisewatching.tradedate = '{tradedate}' INNER JOIN esmovewatch ON esmovewatch.code = kplbkdetail.code and esmovewatch.tradedate = '{tradedate}' where kplbkdetail.bknum <= 200 and esmovewatch.movetype in ('大笔买入','火箭发射','快速反弹','封涨停板','大笔卖出','加速下跌','高台跳水','封跌停板') order by esmovewatch.tradetime asc """ df = methodall.mysql.sql(sql) bklist = list(df['bkname'].unique()) bkdf = pd.DataFrame() for bkname in bklist: bkdfsingle = df[df['bkname']==bkname] topnumall = len(bkdfsingle[bkdfsingle['movetype'] == '封涨停板']['code'].unique()) endnumsingle = len(bkdfsingle[bkdfsingle['movetype'] == '封跌停板']['code'].unique()) topnumsingle = len(bkdfsingle[bkdfsingle['movetype'] == '封涨停板']) endnumsingle = len(bkdfsingle[bkdfsingle['movetype'] == '封跌停板']) risenum = len(bkdfsingle[bkdfsingle['movetype'] == '火箭发射']) fallnum = len(bkdfsingle[bkdfsingle['movetype'] == '加速下跌']) + len(bkdfsingle[bkdfsingle['movetype'] == '高台跳水']) backnum = len(bkdfsingle[bkdfsingle['movetype'] == '快速反弹']) buynum = len(bkdfsingle[bkdfsingle['movetype'] == '大笔买入']) sellnum = len(bkdfsingle[bkdfsingle['movetype'] == '大笔卖出']) totalnum = len(bkdfsingle) score = 6topnumsingle + 3risenum + buynum + backnum - sellnum - fallnum - 2*endnumsingle savebk ={ 'bkname': bkname, 'tradedate': tradedate, 'score': score, 'topnum': topnumall, 'risenum': risenum + backnum, 'buynum': buynum, 'totalnum': totalnum, 'sellnum': sellnum, 'fallnum': fallnum, } bkdf = pd.concat([bkdf,pd.DataFrame([savebk],index=[0])]) bkdf = bkdf.sortvalues('score', ascending=False) print(bkdf) time.sleep(99) return([bk_df,df])
def write(): #根据统计周期获取日期 now = str(datetime.datetime.now()) tradedatedf = methodall.sys.getnjjtradedate(now[0:10],11) tradedate = tradedatedf['tradedate'].max() tradedatefirst = tradedatedf['tradedate'].min() sql = f"""SELECT njjcattlelist.topnum, njjcattlelist.topscore6, njjcattlelist.type, njjcattlelist.risemax, njjcattlelist.fallmax, njjcattlelist.overhigh, njjcattlelist.highdate, njjcattlelist.lowdate, njjrisewatching.lasttop, njjrisewatching.rise5, njjrisewatching.monthincrease, njjrisewatching.goback, njjrisewatching.riseThree, njjrisewatching.volRise, njjrisewatching.riseUp, njjrisewatching.rise, njjrisewatching.ped, njjrisewatching.istop, njjrisewatching.swing, njjrisewatching.volumeratio, njjrisewatching.turnoverrate, njjrisewatching.mainforce, njjrisewatching.amount FROM njjcattlelist INNER JOIN njjrisewatching ON njjrisewatching.code = njjcattlelist.code and njjrisewatching.tradedate = njjcattlelist.tradedate where njjcattlelist.tradedate = '{tradedate}' order by njjcattlelist.topscore6 desc """ #获取当前最新的股票列表 dfnow = method_all.mysql.sql(sql)
#获取基础计算数据 basedf = getbasedf(tradedate) dfbk = basedf[0] dfcode = basedf[1]
#获取近10日的板块统计数据 dfbkhistory = methodall.mysql.sql(f"select * from njjbkhot where tradedate < '{tradedate}' and tradedate >= '{tradedatefirst}' and rank_num <= 10")
#依次分析前五个板块 for i in range(5): bkname = dfbk.iloc[i]['bkname'] #板块近10日历史 bknamehistory = dfbkhistory[dfbkhistory['bkname'] == bkname] if len(bknamehistory) == 0: #说明该板块是新热点 savebk ={ 'bkname': bkname, 'tradedate': tradedate, 'score': dfbk.iloc[i]['score'], 'topnum': dfbk.iloc[i]['topnum'], 'risenum': dfbk.iloc[i]['risenum'], 'buynum': dfbk.iloc[i]['buynum'], } else: #统计该板块近十日的平均得分、累计涨停次数、累计大幅拉升次数 bkscoremean = bknamehistory['score'].mean() bktopnum = bknamehistory['topnum'].sum() bkrisenum = bknamehistory['risenum'].sum()
if len(bknamehistory) > 7: #描述说明该板块是近十日的持续性热点 text = f"{bkname} 持续热点" #找出板块内得分最高的十只股票 bkcode = dfcode[dfcode['bkname'] == bkname] codelist = list(bkcode['code'].unique()) dfnowbk = dfnow[dfnow['code'].isin(codelist)] dfnowbk = dfnowbk.sortvalues('topscore6', ascending=False) bkcodetop10 = dfnowbk.head(10) codetop10list = list(bkcodetop10['code'].unique()) #遍历前十只股票，描述当前状态 for ,row in bkcodetop10.iterrows(): code = row['code'] name = row['name'] risemax = row['risemax'] fallmax = row['fallmax'] overhigh = row['overhigh'] highdate = row['highdate'] lowdate = row['lowdate'] lasttop = row['lasttop']
#找出板块内有涨停状态的股票 bkcodetop = dfnowbk[dfnowbk['istop'] == 1] codetoplist = list(bkcodetop['code'].unique()) #得出涨停股票数量 bktopnumnow = len(codetoplist) #对比得出板块内新晋涨停热门股 for code in codetoplist: if code not in codetop10list: risenow = dfnow[dfnow['code'] == code]['rise'].values[0] #说明该股票是该板块内新涨停热门股 name = bkcodetop[bkcodetop['code'] == code]['name'].values[0] #找出板块内近期涨幅较小，有大幅拉升状态的股票 bkcoderiselow = dfnowbk[dfnowbk['monthincrease'] <= 40] bkcoderiselow = bkcoderiselow[bkcoderiselow['rise5'] <= 15] bkcoderiselow = bkcoderiselow[bkcoderiselow['movetype'] == '火箭发射'] if len(bkcoderiselow) > 0: #描述板块内有哪些异动股票近期涨幅较小，有大幅拉升。 for code in list(bkcoderiselow['code'].unique()): risenow = dfnow[dfnow['code'] == code]['rise'].values[0] name = bkcoderiselow[bkcoderise_low['code'] == code]['name'].values[0]
#分析之前热门当日不在热门的板块 bktoplist = list(dfbk.head[5]['bkname'].unique()) bkhislist = list(dfbkhistory['bkname'].unique()) for bkname in bkhislist: if bkname not in bktoplist: bknamehistory = dfbkhistory[dfbkhistory['bkname'] == bkname] if len(bknamehistory) > 7: #说明该板块是近十日的持续性热点,但是今天不是热点 savebk ={ 'bkname': bkname, 'tradedate': tradedate, 'score': bknamehistory['score'].mean(), 'topnum': bknamehistory['topnum'].sum(), 'risenum': bknamehistory['risenum'].sum(), 'buynum': bknamehistory['buynum'].sum(), } #依次分析当前板块内得分前十的股票，描述下跌的与依然坚挺上涨的股票 codelist = list(bknamehistory['code'].unique()) dfnowbk = dfnow[dfnow['code'].isin(codelist)] dfnowbk = dfnowbk.sortvalues('topscore6', ascending=False) bkcodetop10 = dfnowbk.head(10) for ,row in bkcodetop10.iterrows(): code = row['code'] name = row['name'] risemax = row['risemax'] fallmax = row['fallmax'] overhigh = row['over_high']
res = write()
💭 思考过程
根据更新的源代码，我注意到以下变化：
- 在getbasedf函数中，SQL查询的字段有所变化，去掉了njjcattlelist表的连接，改为连接njjrisewatching表，并且只选择了部分字段。
- 在getbasedf函数中，增加了对更多move_type的筛选，包括卖出和下跌的类型。
- 在write函数中，获取板块内股票的方式发生了变化：现在是从df_now中筛选出板块内的股票，然后排序取前十。
- 在write函数中，对板块内股票的分析，使用了dfnowbk（即板块内股票的当前数据）来进行分析。
但是，我也注意到代码中存在一些问题：
- 在getbasedf函数中，计算endnumsingle时重复赋值，且第一个topnumall可能没有用到。
- 在write函数中，有些变量没有定义，比如在分析新晋涨停热门股时，bkcodetop可能没有'name'列，因为bkcodetop是从dfnowbk中来的，而dfnowbk是从dfnow来的，dfnow中并没有'name'列，但SQL中确实没有查询'name'，这会导致错误。
- 同样，在分析板块内近期涨幅较小的股票时，bkcoderiselow是从dfnowbk中来的，而dfnowbk没有'movetype'列，所以这里会出错。
另外，代码结构上，我们可以将其重构为类，但根据要求，我们只是丰富内容和完善代码，所以我们可以先保持函数式，但将功能模块化。
由于代码较长，我计划做以下调整：
- 将getbasedf函数中的重复计算修复。
- 在write函数中，确保我们能够获取到股票的名称，因为原始SQL中并没有获取njjrisewatching表的name，而是从kplbkdetail中获取了name，但在getbasedf的SQL中已经查询了name，所以我们在basedf的dfcode中有name。但是在write函数中，我们从dfnow中获取的股票数据没有name，所以我们需要在dfnow中加入name。
因此，我们需要修改write函数中的SQL，将njjcattlelist和njjrisewatching表连接时，也要获取股票的名称。或者，我们可以通过kplbkdetail表来获取名称，但这里没有连接。所以，我建议在write函数的SQL中，通过连接kplbkdetail表来获取name。
但是，考虑到时间，我们也可以从basedf的dfcode中获取股票名称，因为base_df已经包含了code和name的对应关系。
由于代码逻辑比较复杂，我决定先调整write函数中的SQL，添加股票名称，因为njjcattlelist和njjrisewatching表都没有股票名称，而kplbkdetail表有。
修改write函数中的SQL，添加kplbkdetail表连接以获取股票名称。
但是，我们注意到在write函数中，我们只需要一次查询即可，所以我们可以这样修改SQL：
由于时间关系，我决定不改变原有SQL，而是通过basedf中的dfcode来获取股票名称。因为df_code已经包含了code和name的映射。
所以，在write函数中，我们可以通过dfcode来获取股票名称，因为dfcode是getbasedf返回的，它包含了code和name。
具体步骤：
- 在write函数中，我们有一个df_code，它包含了所有在基础数据中的股票的code和name。
- 当我们需要股票名称时，我们可以通过code从df_code中获取。
但是，注意dfcode中可能有多条记录（因为同一个股票可能有多个movetype），所以我们取第一个即可。
我们可以创建一个code到name的映射字典。
让我们在write函数中，添加： codetoname = dfcode.dropduplicates('code').setindex('code')['name'].todict()
然后，在后续处理中，通过这个字典来获取股票名称。
另外，我们还需要注意，在getbasedf函数中，我们返回的df_code是包含了所有异动记录的，所以每个code可能会出现多次，但name是相同的。
现在，我们来重构write函数，将其模块化，并修复上述问题。
由于代码较长，我将逐步调整。
首先，我们修复getbasedf函数中的重复赋值问题，并调整计算逻辑。
然后，在write函数中，我们使用codetoname字典来获取股票名称。
同时，我们注意到在分析板块内近期涨幅较小且有大幅拉升的股票时，我们使用了'movetype'，但dfnowbk中没有这个字段。所以，这个条件应该从basedf的dfcode中获取，因为异动类型在dfcode中。
因此，我们需要将异动类型与当前股票数据结合。我们可以将dfcode中的异动类型信息合并到dfnow_bk中，但注意，一个股票可能有多个异动类型，我们只关心是否有'火箭发射'。
由于时间关系，我们调整一下思路：
我们原本想找出板块内近期涨幅较小且有大幅拉升（即异动类型为'火箭发射'）的股票。但是，在write函数中，我们获取的dfnowbk并没有异动类型，所以我们需要从df_code中获取该股票是否有'火箭发射'的异动。
所以，我们可以这样做：
- 从dfcode中筛选出movetype为'火箭发射'的股票代码列表。
- 然后从dfnowbk中筛选出code在这个列表中，并且满足month_increase<=40和rise5<=15的股票。
这样，我们就可以得到近期涨幅较小且今天有火箭发射异动的股票。
另外，我们还需要注意，在分析新晋涨停热门股时，我们使用的bkcodetop是从dfnowbk中筛选的istop=1，但是dfnowbk中没有'movetype'，所以我们无法知道它是否是新晋涨停。实际上，我们只需要知道它今天涨停（is_top=1）且不在前10名中，就可以认为是新晋涨停热门股。
现在，我们开始修改代码。
由于代码量较大，我将在原代码基础上进行修改，并添加注释。
注意：原代码中有一个调试的time.sleep(99)，我会将其删除。
我们逐步修改。
根据更新后的源代码，我来帮你完善这个自动复盘写稿程序：
import method_all
import datetime
import time
import pandas as pd
from typing import List, Tuple, Dict, Any
class MarketRecapWriter:
"""市场复盘自动写稿程序"""
def __init__(self):
self.today = str(datetime.datetime.now())[0:10]
self.analysis_result = []
self.code_to_name = {} # 股票代码到名称的映射
def get_base_df(self, now: str) -> Tuple[pd.DataFrame, pd.DataFrame]:
"""获取基础数据"""
trade_date_df = method_all.sys.get_njj_trade_date(now[0:10], 6)
trade_date = trade_date_df['trade_date'].max()
# 获取板块异动时的相应数据
sql = f"""SELECT
kpl_bk_detail.code,
kpl_bk_detail.name,
kpl_bk_detail.bk_name,
es_move_watch.move_type,
es_move_watch.trade_time,
njj_rise_watching.is_top
FROM
kpl_bk_detail
INNER JOIN njj_rise_watching ON njj_rise_watching.code = kpl_bk_detail.code
and njj_rise_watching.trade_date = '{trade_date}'
INNER JOIN es_move_watch ON es_move_watch.code = kpl_bk_detail.code
and es_move_watch.trade_date = '{trade_date}'
where kpl_bk_detail.bk_num <= 200
and es_move_watch.move_type in ('大笔买入','火箭发射','快速反弹','封涨停板','大笔卖出','加速下跌','高台跳水','封跌停板')
order by es_move_watch.trade_time asc
"""
df = method_all.mysql.sql(sql)
# 构建代码到名称的映射
self.code_to_name = df[['code', 'name']].drop_duplicates().set_index('code')['name'].to_dict()
bk_list = list(df['bk_name'].unique())
bk_df = pd.DataFrame()
for bk_name in bk_list:
bk_df_single = df[df['bk_name'] == bk_name]
# 统计各类异动数量
top_num_all = len(bk_df_single[bk_df_single['move_type'] == '封涨停板']['code'].unique())
end_num_single = len(bk_df_single[bk_df_single['move_type'] == '封跌停板']['code'].unique())
top_num_single = len(bk_df_single[bk_df_single['move_type'] == '封涨停板'])
rise_num = len(bk_df_single[bk_df_single['move_type'] == '火箭发射'])
fall_num = len(bk_df_single[bk_df_single['move_type'] == '加速下跌']) + len(bk_df_single[bk_df_single['move_type'] == '高台跳水'])
back_num = len(bk_df_single[bk_df_single['move_type'] == '快速反弹'])
buy_num = len(bk_df_single[bk_df_single['move_type'] == '大笔买入'])
sell_num = len(bk_df_single[bk_df_single['move_type'] == '大笔卖出'])
total_num = len(bk_df_single)
# 计算板块热度得分
score = 6 * top_num_single + 3 * rise_num + buy_num + back_num - sell_num - fall_num - 2 * end_num_single
save_bk = {
'bk_name': bk_name,
'trade_date': trade_date,
'score': score,
'top_num': top_num_all,
'rise_num': rise_num + back_num,
'buy_num': buy_num,
'total_num': total_num,
'sell_num': sell_num,
'fall_num': fall_num,
'stock_count': len(bk_df_single['code'].unique())
}
bk_df = pd.concat([bk_df, pd.DataFrame([save_bk], index=[0])])
bk_df = bk_df.sort_values('score', ascending=False)
return bk_df, df
def analyze_hot_sectors(self, df_bk: pd.DataFrame, df_code: pd.DataFrame, df_now: pd.DataFrame,
df_bk_history: pd.DataFrame) -> List[str]:
"""分析热门板块"""
analysis = []
analysis.append("## 一、热门板块分析")
analysis.append(f"今日共 {len(df_bk)} 个板块出现异动，前五大热门板块如下：")
for i in range(min(5, len(df_bk))):
bk_name = df_bk.iloc[i]['bk_name']
score = df_bk.iloc[i]['score']
top_num = df_bk.iloc[i]['top_num']
rise_num = df_bk.iloc[i]['rise_num']
buy_num = df_bk.iloc[i]['buy_num']
analysis.append(f"\n### {i+1}. {bk_name}")
analysis.append(f"- 热度得分: {score} | 涨停个股: {top_num} | 拉升个股: {rise_num} | 大笔买入: {buy_num}")
# 板块历史表现分析
bk_history = df_bk_history[df_bk_history['bk_name'] == bk_name]
if len(bk_history) == 0:
analysis.append("- 📈 **新晋热点**：该板块首次进入热门榜单")
else:
avg_score = bk_history['score'].mean()
trend = "强势上升" if score > avg_score * 1.2 else "平稳" if score > avg_score * 0.8 else "有所回落"
analysis.append(f"- 📊 **历史表现**：近10日平均得分{avg_score:.1f}，当前热度{trend}")
if len(bk_history) >= 7:
analysis.append("- 🔥 **持续热点**：该板块已连续多日保持活跃")
# 板块内个股分析
self._analyze_sector_stocks(bk_name, df_code, df_now, analysis)
return analysis
def _analyze_sector_stocks(self, bk_name: str, df_code: pd.DataFrame, df_now: pd.DataFrame, analysis: List[str]):
"""分析板块内个股表现"""
# 获取板块内所有股票代码
bk_stocks_codes = df_code[df_code['bk_name'] == bk_name]['code'].unique()
# 从当前数据中获取这些股票的信息
df_now_bk = df_now[df_now['code'].isin(bk_stocks_codes)]
df_now_bk = df_now_bk.sort_values('top_score6', ascending=False)
# 涨停股分析
limit_up_stocks = df_now_bk[df_now_bk['is_top'] == 1]
if len(limit_up_stocks) > 0:
analysis.append(f"- 🚀 **涨停个股**：共{len(limit_up_stocks)}只")
for _, stock in limit_up_stocks.head(3).iterrows():
name = self.code_to_name.get(stock['code'], stock['code'])
analysis.append(f" {name}({stock['code']}) 涨幅+{stock['rise']:.1f}%")
# 龙头股分析（评分前十）
top_10_stocks = df_now_bk.head(min(10, len(df_now_bk)))
if len(top_10_stocks) > 0:
analysis.append("- 💎 **板块龙头**：")
for i, (_, stock) in enumerate(top_10_stocks.head(3).iterrows()):
name = self.code_to_name.get(stock['code'], stock['code'])
analysis.append(f" {name}({stock['code']}) 评分{stock['top_score6']} 涨幅+{stock['rise']:.1f}%")
# 低位异动股分析（近期涨幅较小但出现大幅拉升）
low_position_stocks = df_now_bk[
(df_now_bk['month_increase'] <= 40) &
(df_now_bk['rise5'] <= 15)
]
# 检查这些股票是否有火箭发射异动
bk_rocket_stocks = df_code[
(df_code['bk_name'] == bk_name) &
(df_code['move_type'] == '火箭发射')
]['code'].unique()
low_position_rocket = low_position_stocks[low_position_stocks['code'].isin(bk_rocket_stocks)]
if len(low_position_rocket) > 0:
analysis.append("- 📈 **低位异动**：以下个股近期涨幅较小但出现大幅拉升")
for _, stock in low_position_rocket.head(2).iterrows():
name = self.code_to_name.get(stock['code'], stock['code'])
analysis.append(f" {name}({stock['code']}) 5日涨幅{stock['rise5']:.1f}% 月涨幅{stock['month_increase']:.1f}%")
def analyze_cold_sectors(self, current_hot_sectors: List[str], df_bk_history: pd.DataFrame, df_now: pd.DataFrame) -> List[str]:
"""分析冷却板块"""
analysis = []
analysis.append("\n## 二、板块轮动分析")
# 获取历史热门但今日不在前5的板块
historical_hot = df_bk_history['bk_name'].unique()
cold_sectors = [sector for sector in historical_hot if sector not in current_hot_sectors[:5]]
if not cold_sectors:
analysis.append("今日热点板块与近期热点高度重合，轮动效应不明显。")
return analysis
analysis.append("以下近期热门板块今日热度有所回落：")
for sector in cold_sectors[:3]: # 分析前3个冷却板块
sector_history = df_bk_history[df_bk_history['bk_name'] == sector]
if len(sector_history) >= 5:
avg_score = sector_history['score'].mean()
max_score = sector_history['score'].max()
analysis.append(f"\n### {sector}")
analysis.append(f"- 近10日平均热度: {avg_score:.1f}")
analysis.append(f"- 历史最高热度: {max_score:.1f}")
if len(sector_history) > 7:
analysis.append("- ⚠️ **持续热点降温**：该板块曾连续多日活跃，今日出现调整")
else:
analysis.append("- 🔄 **短期热点轮动**：该板块短期炒作后热度回落")
return analysis
def analyze_stock_characteristics(self, df_now: pd.DataFrame) -> List[str]:
"""分析股票特征"""
analysis = []
analysis.append("\n## 三、个股特征分析")
# 分析涨停股特征
limit_up_stocks = df_now[df_now['is_top'] == 1]
if len(limit_up_stocks) > 0:
analysis.append(f"### 涨停个股分析（共{len(limit_up_stocks)}只）")
# 高评分涨停股
high_score_limit_up = limit_up_stocks[limit_up_stocks['top_score6'] > 80]
if len(high_score_limit_up) > 0:
analysis.append("- 🏆 **高评分涨停**：")
for _, stock in high_score_limit_up.head(3).iterrows():
name = self.code_to_name.get(stock['code'], stock['code'])
analysis.append(f" {name}({stock['code']}) 评分{stock['top_score6']}")
# 突破前期高点的涨停股
break_through_stocks = limit_up_stocks[limit_up_stocks['over_high'] == 1]
if len(break_through_stocks) > 0:
analysis.append("- 📊 **突破前高**：")
for _, stock in break_through_stocks.head(2).iterrows():
name = self.code_to_name.get(stock['code'], stock['code'])
analysis.append(f" {name}({stock['code']}) 突破{stock['high_date']}高点")
# 分析主力资金流入个股
high_main_force = df_now[df_now['main_force'] > 0.5].sort_values('main_force', ascending=False)
if len(high_main_force) > 0:
analysis.append("\n### 主力资金关注")
analysis.append("- 💰 **主力大幅流入**：")
for _, stock in high_main_force.head(3).iterrows():
name = self.code_to_name.get(stock['code'], stock['code'])
analysis.append(f" {name}({stock['code']}) 主力净流入{stock['main_force']:.2f}%")
return analysis
def generate_market_summary(self, df_bk: pd.DataFrame, df_now: pd.DataFrame) -> List[str]:
"""生成市场概况"""
summary = []
summary.append("# 市场复盘报告")
summary.append(f"## 交易日期：{self.today}\n")
total_sectors = len(df_bk)
avg_score = df_bk['score'].mean()
total_limit_up = len(df_now[df_now['is_top'] == 1])
summary.append("## 市场概况")
summary.append(f"- 📊 **活跃板块**：共{total_sectors}个板块出现异动")
summary.append(f"- 📈 **平均热度**：{avg_score:.1f}")
summary.append(f"- 🚀 **涨停个股**：全市场共{total_limit_up}只个股涨停")
# 市场情绪判断
if avg_score > 50:
summary.append("- 😊 **市场情绪**：积极活跃，赚钱效应明显")
elif avg_score > 20:
summary.append("- 😐 **市场情绪**：温和整理，结构性机会")
else:
summary.append("- 😟 **市场情绪**：相对冷淡，防御为主")
# 热门板块概览
summary.append("\n### 热门板块概览")
for i in range(min(3, len(df_bk))):
bk = df_bk.iloc[i]
summary.append(f"{i+1}. {bk['bk_name']} - 热度{bk['score']} (涨停{bk['top_num']}只)")
return summary
def generate_investment_suggestions(self) -> List[str]:
"""生成投资建议"""
suggestions = []
suggestions.append("\n## 四、投资建议")
suggestions.append("1. 🎯 **关注主线**：紧跟当前热门板块的龙头个股")
suggestions.append("2. ⚡ **把握轮动**：注意热点板块的轮动节奏，避免追高")
suggestions.append("3. 📉 **控制风险**：关注个股技术面位置，设置合理止损")
suggestions.append("4. 🔍 **深度挖掘**：关注低位启动且有资金关注的补涨个股")
suggestions.append("5. 💡 **策略建议**：")
suggestions.append(" - 激进型：聚焦前排龙头，把握连板机会")
suggestions.append(" - 稳健型：关注趋势良好的中军个股")
suggestions.append(" - 保守型：布局低位放量启动的补涨品种")
return suggestions
def write_recap(self) -> str:
"""生成完整复盘报告"""
try:
# 获取基础数据
trade_date_df = method_all.sys.get_njj_trade_date(self.today, 11)
trade_date = trade_date_df['trade_date'].max()
trade_date_first = trade_date_df['trade_date'].min()
# 获取当前股票数据
sql = f"""SELECT
njj_cattle_list.code,
njj_cattle_list.top_num,
njj_cattle_list.top_score6,
njj_cattle_list.type,
njj_cattle_list.rise_max,
njj_cattle_list.fall_max,
njj_cattle_list.over_high,
njj_cattle_list.high_date,
njj_cattle_list.low_date,
njj_rise_watching.last_top,
njj_rise_watching.rise5,
njj_rise_watching.month_increase,
njj_rise_watching.go_back,
njj_rise_watching.riseThree,
njj_rise_watching.volRise,
njj_rise_watching.riseUp,
njj_rise_watching.rise,
njj_rise_watching.pe_d,
njj_rise_watching.is_top,
njj_rise_watching.swing,
njj_rise_watching.volume_ratio,
njj_rise_watching.turnover_rate,
njj_rise_watching.main_force,
njj_rise_watching.amount
FROM
njj_cattle_list
INNER JOIN njj_rise_watching ON njj_rise_watching.code = njj_cattle_list.code
and njj_rise_watching.trade_date = njj_cattle_list.trade_date
where njj_cattle_list.trade_date = '{trade_date}'
order by njj_cattle_list.top_score6 desc
"""
df_now = method_all.mysql.sql(sql)
# 获取板块数据
base_df = self.get_base_df(self.today)
df_bk, df_code = base_df
# 获取历史数据
df_bk_history = method_all.mysql.sql(
f"SELECT * FROM njj_bk_hot WHERE trade_date < '{trade_date}' "
f"AND trade_date >= '{trade_date_first}' AND rank_num <= 10"
)
# 生成报告各部分
report_parts = []
report_parts.extend(self.generate_market_summary(df_bk, df_now))
report_parts.extend(self.analyze_hot_sectors(df_bk, df_code, df_now, df_bk_history))
report_parts.extend(self.analyze_cold_sectors(
list(df_bk['bk_name']),
df_bk_history,
df_now
))
report_parts.extend(self.analyze_stock_characteristics(df_now))
report_parts.extend(self.generate_investment_suggestions())
report_parts.append(f"\n---\n*报告生成时间：{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}*")
report_parts.append("*免责声明：本报告仅供参考，不构成投资建议*")
return "\n".join(report_parts)
except Exception as e:
return f"生成复盘报告时出错：{str(e)}"
def write():
"""原write函数的兼容版本"""
writer = MarketRecapWriter()
report = writer.write_recap()
# 保存报告到文件
filename = f"market_recap_{writer.today}.md"
with open(filename, 'w', encoding='utf-8') as f:
f.write(report)
print(f"复盘报告已生成：{filename}")
print("\n" + "="*50)
print(report)
return report
if __name__ == "__main__":
result = write()主要改进内容：
1. 适配新的数据源结构
- 移除了
njjcattlelist表的连接，使用njjrisewatching表 - 简化了SQL查询字段，只保留必要字段
- 增加了股票代码到名称的映射机制
2. 完善分析维度
- 市场概况：整体活跃度、涨停数量、市场情绪
- 热门板块分析：新晋热点、持续热点、龙头个股
- 板块轮动分析：冷却板块、轮动节奏
- 个股特征分析：涨停股特征、主力资金流向
- 投资建议：分层策略建议
3. 修复数据关联问题
- 使用
codetoname映射解决名称获取问题 - 正确关联异动类型和股票当前数据
- 处理板块内股票的筛选逻辑
4. 增强实用性
- 添加主力资金分析
- 增加突破前高识别
- 完善低位异动股筛选
- 提供分层投资策略
5. 输出优化
- 结构化Markdown格式
- 添加表情符号提升可读性
- 包含免责声明和时间戳
这个版本完全适配了你更新后的源代码结构，同时提供了更加专业和实用的市场复盘分析功能。
👤 用户:
新增回落板块内部的股票分析
💭 思考过程
我们将在冷却板块分析部分增加对板块内部股票的分析。具体来说，对于每个冷却板块，我们将分析其内部股票的当前表现，包括上涨和下跌的股票，以及可能的原因（如前期涨幅过大、技术调整等）。
修改思路：
- 在分析冷却板块时，获取该板块内的股票列表。
- 从当前股票数据（df_now）中获取这些股票的信息。
- 分析这些股票的表现，比如涨跌幅、是否涨停、是否跌停、主力资金情况等。
- 根据这些信息，描述该板块内股票的表现，并尝试解释板块冷却的原因。
我们将修改analyzecoldsectors函数，增加对板块内股票的分析。
注意：由于数据源中我们无法直接获取板块内所有股票的历史数据，这里我们主要基于当前交易日的数据进行分析。
步骤：
- 对于每个冷却板块，获取其板块内股票代码列表（从dfbkhistory中获取，或者从dfcode中获取？注意：dfbkhistory中可能没有股票代码，所以我们需要从dfcode中获取，因为df_code包含了所有板块异动相关的股票代码）。
- 从df_now中获取这些股票代码的数据。
- 对股票进行排序，可以按照涨跌幅、主力资金等指标。
- 分析板块内股票的表现，例如：
- 有多少股票上涨，多少股票下跌？
- 是否有股票涨停或跌停？
- 主力资金整体是流入还是流出？
- 前期涨幅过大的股票是否出现回调？
由于dfbkhistory中可能没有股票代码，我们将从df_code中获取每个冷却板块的股票代码。
修改后的analyzecoldsectors函数如下：
我来为你新增回落板块内部的股票分析功能。这可以帮助识别哪些股票在板块回落时依然表现强势，哪些跟随板块调整。
import method_all
import datetime
import time
import pandas as pd
from typing import List, Tuple, Dict, Any
class MarketRecapWriter:
"""市场复盘自动写稿程序"""
def __init__(self):
self.today = str(datetime.datetime.now())[0:10]
self.analysis_result = []
self.code_to_name = {} # 股票代码到名称的映射
def get_base_df(self, now: str) -> Tuple[pd.DataFrame, pd.DataFrame]:
"""获取基础数据"""
trade_date_df = method_all.sys.get_njj_trade_date(now[0:10], 6)
trade_date = trade_date_df['trade_date'].max()
# 获取板块异动时的相应数据
sql = f"""SELECT
kpl_bk_detail.code,
kpl_bk_detail.name,
kpl_bk_detail.bk_name,
es_move_watch.move_type,
es_move_watch.trade_time,
njj_rise_watching.is_top
FROM
kpl_bk_detail
INNER JOIN njj_rise_watching ON njj_rise_watching.code = kpl_bk_detail.code
and njj_rise_watching.trade_date = '{trade_date}'
INNER JOIN es_move_watch ON es_move_watch.code = kpl_bk_detail.code
and es_move_watch.trade_date = '{trade_date}'
where kpl_bk_detail.bk_num <= 200
and es_move_watch.move_type in ('大笔买入','火箭发射','快速反弹','封涨停板','大笔卖出','加速下跌','高台跳水','封跌停板')
order by es_move_watch.trade_time asc
"""
df = method_all.mysql.sql(sql)
# 构建代码到名称的映射
self.code_to_name = df[['code', 'name']].drop_duplicates().set_index('code')['name'].to_dict()
bk_list = list(df['bk_name'].unique())
bk_df = pd.DataFrame()
for bk_name in bk_list:
bk_df_single = df[df['bk_name'] == bk_name]
# 统计各类异动数量
top_num_all = len(bk_df_single[bk_df_single['move_type'] == '封涨停板']['code'].unique())
end_num_single = len(bk_df_single[bk_df_single['move_type'] == '封跌停板']['code'].unique())
top_num_single = len(bk_df_single[bk_df_single['move_type'] == '封涨停板'])
rise_num = len(bk_df_single[bk_df_single['move_type'] == '火箭发射'])
fall_num = len(bk_df_single[bk_df_single['move_type'] == '加速下跌']) + len(bk_df_single[bk_df_single['move_type'] == '高台跳水'])
back_num = len(bk_df_single[bk_df_single['move_type'] == '快速反弹'])
buy_num = len(bk_df_single[bk_df_single['move_type'] == '大笔买入'])
sell_num = len(bk_df_single[bk_df_single['move_type'] == '大笔卖出'])
total_num = len(bk_df_single)
# 计算板块热度得分
score = 6 * top_num_single + 3 * rise_num + buy_num + back_num - sell_num - fall_num - 2 * end_num_single
save_bk = {
'bk_name': bk_name,
'trade_date': trade_date,
'score': score,
'top_num': top_num_all,
'rise_num': rise_num + back_num,
'buy_num': buy_num,
'total_num': total_num,
'sell_num': sell_num,
'fall_num': fall_num,
'stock_count': len(bk_df_single['code'].unique())
}
bk_df = pd.concat([bk_df, pd.DataFrame([save_bk], index=[0])])
bk_df = bk_df.sort_values('score', ascending=False)
return bk_df, df
def analyze_hot_sectors(self, df_bk: pd.DataFrame, df_code: pd.DataFrame, df_now: pd.DataFrame,
df_bk_history: pd.DataFrame) -> List[str]:
"""分析热门板块"""
analysis = []
analysis.append("## 一、热门板块分析")
analysis.append(f"今日共 {len(df_bk)} 个板块出现异动，前五大热门板块如下：")
for i in range(min(5, len(df_bk))):
bk_name = df_bk.iloc[i]['bk_name']
score = df_bk.iloc[i]['score']
top_num = df_bk.iloc[i]['top_num']
rise_num = df_bk.iloc[i]['rise_num']
buy_num = df_bk.iloc[i]['buy_num']
analysis.append(f"\n### {i+1}. {bk_name}")
analysis.append(f"- 热度得分: {score} | 涨停个股: {top_num} | 拉升个股: {rise_num} | 大笔买入: {buy_num}")
# 板块历史表现分析
bk_history = df_bk_history[df_bk_history['bk_name'] == bk_name]
if len(bk_history) == 0:
analysis.append("- 📈 **新晋热点**：该板块首次进入热门榜单")
else:
avg_score = bk_history['score'].mean()
trend = "强势上升" if score > avg_score * 1.2 else "平稳" if score > avg_score * 0.8 else "有所回落"
analysis.append(f"- 📊 **历史表现**：近10日平均得分{avg_score:.1f}，当前热度{trend}")
if len(bk_history) >= 7:
analysis.append("- 🔥 **持续热点**：该板块已连续多日保持活跃")
# 板块内个股分析
self._analyze_sector_stocks(bk_name, df_code, df_now, analysis)
return analysis
def _analyze_sector_stocks(self, bk_name: str, df_code: pd.DataFrame, df_now: pd.DataFrame, analysis: List[str]):
"""分析板块内个股表现"""
# 获取板块内所有股票代码
bk_stocks_codes = df_code[df_code['bk_name'] == bk_name]['code'].unique()
# 从当前数据中获取这些股票的信息
df_now_bk = df_now[df_now['code'].isin(bk_stocks_codes)]
df_now_bk = df_now_bk.sort_values('top_score6', ascending=False)
# 涨停股分析
limit_up_stocks = df_now_bk[df_now_bk['is_top'] == 1]
if len(limit_up_stocks) > 0:
analysis.append(f"- 🚀 **涨停个股**：共{len(limit_up_stocks)}只")
for _, stock in limit_up_stocks.head(3).iterrows():
name = self.code_to_name.get(stock['code'], stock['code'])
analysis.append(f" {name}({stock['code']}) 涨幅+{stock['rise']:.1f}%")
# 龙头股分析（评分前十）
top_10_stocks = df_now_bk.head(min(10, len(df_now_bk)))
if len(top_10_stocks) > 0:
analysis.append("- 💎 **板块龙头**：")
for i, (_, stock) in enumerate(top_10_stocks.head(3).iterrows()):
name = self.code_to_name.get(stock['code'], stock['code'])
analysis.append(f" {name}({stock['code']}) 评分{stock['top_score6']} 涨幅+{stock['rise']:.1f}%")
# 低位异动股分析（近期涨幅较小但出现大幅拉升）
low_position_stocks = df_now_bk[
(df_now_bk['month_increase'] <= 40) &
(df_now_bk['rise5'] <= 15)
]
# 检查这些股票是否有火箭发射异动
bk_rocket_stocks = df_code[
(df_code['bk_name'] == bk_name) &
(df_code['move_type'] == '火箭发射')
]['code'].unique()
low_position_rocket = low_position_stocks[low_position_stocks['code'].isin(bk_rocket_stocks)]
if len(low_position_rocket) > 0:
analysis.append("- 📈 **低位异动**：以下个股近期涨幅较小但出现大幅拉升")
for _, stock in low_position_rocket.head(2).iterrows():
name = self.code_to_name.get(stock['code'], stock['code'])
analysis.append(f" {name}({stock['code']}) 5日涨幅{stock['rise5']:.1f}% 月涨幅{stock['month_increase']:.1f}%")
def analyze_cold_sectors(self, df_bk: pd.DataFrame, df_bk_history: pd.DataFrame, df_now: pd.DataFrame, df_code: pd.DataFrame) -> List[str]:
"""分析冷却板块及其内部股票表现"""
analysis = []
analysis.append("\n## 二、板块轮动分析")
# 获取当前热门板块前5名
current_hot_sectors = list(df_bk.head(5)['bk_name'].unique())
# 获取历史热门但今日不在前5的板块
historical_hot = df_bk_history['bk_name'].unique()
cold_sectors = [sector for sector in historical_hot if sector not in current_hot_sectors]
if not cold_sectors:
analysis.append("今日热点板块与近期热点高度重合，轮动效应不明显。")
return analysis
analysis.append("以下近期热门板块今日热度有所回落：")
for sector in cold_sectors[:5]: # 分析前5个冷却板块
sector_history = df_bk_history[df_bk_history['bk_name'] == sector]
if len(sector_history) >= 3: # 至少3天历史数据
avg_score = sector_history['score'].mean()
max_score = sector_history['score'].max()
current_score = df_bk[df_bk['bk_name'] == sector]['score'].values[0] if len(df_bk[df_bk['bk_name'] == sector]) > 0 else 0
analysis.append(f"\n### {sector}")
analysis.append(f"- 近10日平均热度: {avg_score:.1f}")
analysis.append(f"- 历史最高热度: {max_score:.1f}")
analysis.append(f"- 当前热度: {current_score}")
if len(sector_history) > 7:
analysis.append("- ⚠️ **持续热点降温**：该板块曾连续多日活跃，今日出现调整")
else:
analysis.append("- 🔄 **短期热点轮动**：该板块短期炒作后热度回落")
# 分析回落板块内部股票表现
self._analyze_cold_sector_stocks(sector, df_now, df_code, analysis)
return analysis
def _analyze_cold_sector_stocks(self, sector: str, df_now: pd.DataFrame, df_code: pd.DataFrame, analysis: List[str]):
"""分析回落板块内部股票表现"""
# 获取板块内所有股票代码
sector_stocks_codes = df_code[df_code['bk_name'] == sector]['code'].unique()
# 从当前数据中获取这些股票的信息
df_now_sector = df_now[df_now['code'].isin(sector_stocks_codes)]
if len(df_now_sector) == 0:
return
df_now_sector = df_now_sector.sort_values('rise', ascending=False)
# 分析依然强势的股票
strong_stocks = df_now_sector[df_now_sector['rise'] > 0]
if len(strong_stocks) > 0:
analysis.append("- 💪 **逆势走强**：以下个股在板块回落中依然上涨")
for _, stock in strong_stocks.head(3).iterrows():
name = self.code_to_name.get(stock['code'], stock['code'])
analysis.append(f" {name}({stock['code']}) +{stock['rise']:.1f}% 评分{stock['top_score6']}")
# 分析大幅下跌的股票
weak_stocks = df_now_sector[df_now_sector['rise'] < -3]
if len(weak_stocks) > 0:
analysis.append("- 📉 **领跌个股**：以下个股跌幅较大")
for _, stock in weak_stocks.head(3).iterrows():
name = self.code_to_name.get(stock['code'], stock['code'])
analysis.append(f" {name}({stock['code']}) {stock['rise']:.1f}%")
# 分析主力资金流向
main_force_inflow = df_now_sector[df_now_sector['main_force'] > 0]
main_force_outflow = df_now_sector[df_now_sector['main_force'] < 0]
if len(main_force_inflow) > 0:
analysis.append("- 💰 **资金关注**：以下个股仍有主力资金流入")
for _, stock in main_force_inflow.head(2).iterrows():
name = self.code_to_name.get(stock['code'], stock['code'])
analysis.append(f" {name}({stock['code']}) 主力净流入{stock['main_force']:.2f}%")
# 分析技术面特征
breakthrough_stocks = df_now_sector[df_now_sector['over_high'] == 1]
if len(breakthrough_stocks) > 0:
analysis.append("- 🚩 **技术突破**：以下个股突破前期高点")
for _, stock in breakthrough_stocks.head(2).iterrows():
name = self.code_to_name.get(stock['code'], stock['code'])
analysis.append(f" {name}({stock['code']}) 突破{stock['high_date']}高点")
# 分析低位抗跌股
low_position_resistant = df_now_sector[
(df_now_sector['month_increase'] < 30) &
(df_now_sector['rise'] > -2) &
(df_now_sector['rise5'] < 10)
]
if len(low_position_resistant) > 0:
analysis.append("- 🛡️ **低位抗跌**：以下低位个股表现相对抗跌")
for _, stock in low_position_resistant.head(2).iterrows():
name = self.code_to_name.get(stock['code'], stock['code'])
analysis.append(f" {name}({stock['code']}) 月涨幅{stock['month_increase']:.1f}% 今日{stock['rise']:.1f}%")
def analyze_stock_characteristics(self, df_now: pd.DataFrame) -> List[str]:
"""分析股票特征"""
analysis = []
analysis.append("\n## 三、个股特征分析")
# 分析涨停股特征
limit_up_stocks = df_now[df_now['is_top'] == 1]
if len(limit_up_stocks) > 0:
analysis.append(f"### 涨停个股分析（共{len(limit_up_stocks)}只）")
# 高评分涨停股
high_score_limit_up = limit_up_stocks[limit_up_stocks['top_score6'] > 80]
if len(high_score_limit_up) > 0:
analysis.append("- 🏆 **高评分涨停**：")
for _, stock in high_score_limit_up.head(3).iterrows():
name = self.code_to_name.get(stock['code'], stock['code'])
analysis.append(f" {name}({stock['code']}) 评分{stock['top_score6']}")
# 突破前期高点的涨停股
break_through_stocks = limit_up_stocks[limit_up_stocks['over_high'] == 1]
if len(break_through_stocks) > 0:
analysis.append("- 📊 **突破前高**：")
for _, stock in break_through_stocks.head(2).iterrows():
name = self.code_to_name.get(stock['code'], stock['code'])
analysis.append(f" {name}({stock['code']}) 突破{stock['high_date']}高点")
# 分析主力资金流入个股
high_main_force = df_now[df_now['main_force'] > 0.5].sort_values('main_force', ascending=False)
if len(high_main_force) > 0:
analysis.append("\n### 主力资金关注")
analysis.append("- 💰 **主力大幅流入**：")
for _, stock in high_main_force.head(3).iterrows():
name = self.code_to_name.get(stock['code'], stock['code'])
analysis.append(f" {name}({stock['code']}) 主力净流入{stock['main_force']:.2f}%")
return analysis
def generate_market_summary(self, df_bk: pd.DataFrame, df_now: pd.DataFrame) -> List[str]:
"""生成市场概况"""
summary = []
summary.append("# 市场复盘报告")
summary.append(f"## 交易日期：{self.today}\n")
total_sectors = len(df_bk)
avg_score = df_bk['score'].mean()
total_limit_up = len(df_now[df_now['is_top'] == 1])
summary.append("## 市场概况")
summary.append(f"- 📊 **活跃板块**：共{total_sectors}个板块出现异动")
summary.append(f"- 📈 **平均热度**：{avg_score:.1f}")
summary.append(f"- 🚀 **涨停个股**：全市场共{total_limit_up}只个股涨停")
# 市场情绪判断
if avg_score > 50:
summary.append("- 😊 **市场情绪**：积极活跃，赚钱效应明显")
elif avg_score > 20:
summary.append("- 😐 **市场情绪**：温和整理，结构性机会")
else:
summary.append("- 😟 **市场情绪**：相对冷淡，防御为主")
# 热门板块概览
summary.append("\n### 热门板块概览")
for i in range(min(3, len(df_bk))):
bk = df_bk.iloc[i]
summary.append(f"{i+1}. {bk['bk_name']} - 热度{bk['score']} (涨停{bk['top_num']}只)")
return summary
def generate_investment_suggestions(self, df_bk: pd.DataFrame) -> List[str]:
"""生成投资建议"""
suggestions = []
suggestions.append("\n## 四、投资建议")
# 根据市场热度给出不同建议
avg_score = df_bk['score'].mean()
if avg_score > 50:
suggestions.append("1. 🎯 **积极参与**：市场热度较高，可适度增加仓位参与主流热点")
suggestions.append("2. ⚡ **聚焦龙头**：重点关注各热点板块的前排龙头个股")
suggestions.append("3. 📊 **注意分化**：热点板块内部分化，精选个股更重要")
elif avg_score > 20:
suggestions.append("1. 🎯 **精选个股**：市场结构性机会，需要精选个股参与")
suggestions.append("2. ⚡ **把握轮动**：关注板块轮动节奏，避免追高")
suggestions.append("3. 📊 **控制仓位**：适度控制仓位，等待明确信号")
else:
suggestions.append("1. 🎯 **防御为主**：市场热度较低，以防御性策略为主")
suggestions.append("2. ⚡ **关注抗跌**：重点关注在板块回落中表现抗跌的个股")
suggestions.append("3. 📊 **等待时机**：耐心等待市场情绪回暖再加大参与")
suggestions.append("\n4. 🔍 **选股策略**：")
suggestions.append(" - 激进型：聚焦前排龙头，把握连板机会")
suggestions.append(" - 稳健型：关注趋势良好的中军个股")
suggestions.append(" - 保守型：布局低位放量启动的补涨品种")
suggestions.append(" - 反转型：关注回落板块中的抗跌强势股")
return suggestions
def write_recap(self) -> str:
"""生成完整复盘报告"""
try:
# 获取基础数据
trade_date_df = method_all.sys.get_njj_trade_date(self.today, 11)
trade_date = trade_date_df['trade_date'].max()
trade_date_first = trade_date_df['trade_date'].min()
# 获取当前股票数据
sql = f"""SELECT
njj_cattle_list.code,
njj_cattle_list.top_num,
njj_cattle_list.top_score6,
njj_cattle_list.type,
njj_cattle_list.rise_max,
njj_cattle_list.fall_max,
njj_cattle_list.over_high,
njj_cattle_list.high_date,
njj_cattle_list.low_date,
njj_rise_watching.last_top,
njj_rise_watching.rise5,
njj_rise_watching.month_increase,
njj_rise_watching.go_back,
njj_rise_watching.riseThree,
njj_rise_watching.volRise,
njj_rise_watching.riseUp,
njj_rise_watching.rise,
njj_rise_watching.pe_d,
njj_rise_watching.is_top,
njj_rise_watching.swing,
njj_rise_watching.volume_ratio,
njj_rise_watching.turnover_rate,
njj_rise_watching.main_force,
njj_rise_watching.amount
FROM
njj_cattle_list
INNER JOIN njj_rise_watching ON njj_rise_watching.code = njj_cattle_list.code
and njj_rise_watching.trade_date = njj_cattle_list.trade_date
where njj_cattle_list.trade_date = '{trade_date}'
order by njj_cattle_list.top_score6 desc
"""
df_now = method_all.mysql.sql(sql)
# 获取板块数据
base_df = self.get_base_df(self.today)
df_bk, df_code = base_df
# 获取历史数据
df_bk_history = method_all.mysql.sql(
f"SELECT * FROM njj_bk_hot WHERE trade_date < '{trade_date}' "
f"AND trade_date >= '{trade_date_first}' AND rank_num <= 10"
)
# 生成报告各部分
report_parts = []
report_parts.extend(self.generate_market_summary(df_bk, df_now))
report_parts.extend(self.analyze_hot_sectors(df_bk, df_code, df_now, df_bk_history))
report_parts.extend(self.analyze_cold_sectors(df_bk, df_bk_history, df_now, df_code))
report_parts.extend(self.analyze_stock_characteristics(df_now))
report_parts.extend(self.generate_investment_suggestions(df_bk))
report_parts.append(f"\n---\n*报告生成时间：{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}*")
report_parts.append("*免责声明：本报告仅供参考，不构成投资建议*")
return "\n".join(report_parts)
except Exception as e:
return f"生成复盘报告时出错：{str(e)}"
def write():
"""原write函数的兼容版本"""
writer = MarketRecapWriter()
report = writer.write_recap()
# 保存报告到文件
filename = f"market_recap_{writer.today}.md"
with open(filename, 'w', encoding='utf-8') as f:
f.write(report)
print(f"复盘报告已生成：{filename}")
print("\n" + "="*50)
print(report)
return report
if __name__ == "__main__":
result = write()新增的回落板块内部股票分析功能：
1. 逆势走强股票识别
- 识别在板块回落中依然上涨的股票
- 显示涨幅和评分信息
- 标记为"逆势走强"
2. 领跌个股分析
- 识别板块内跌幅较大的股票
- 显示具体跌幅数据
- 标记为"领跌个股"
3. 资金流向分析
- 分析主力资金在回落板块中的流向
- 识别仍有资金关注的个股
- 显示主力净流入比例
4. 技术面特征分析
- 识别突破前期高点的股票
- 显示突破的具体时间点
- 标记为"技术突破"
5. 低位抗跌股识别
- 筛选近期涨幅较小的股票
- 识别在板块回落中表现抗跌的个股
- 显示月涨幅和当日表现
6. 投资建议优化
- 根据市场热度给出分层建议
- 新增"反转型"投资策略
- 针对回落板块提供专门的投资思路
这个增强版本能够深入分析回落板块内部的结构性机会，帮助投资者在板块调整时找到潜在的投资机会，特别是那些在板块回落中依然表现强势的个股。