from datetime import timedelta import numpy as np import pandas from utils.comm import * class AllIndex: # 中国的指数 ZH_INDEX = pd.DataFrame( [["上证指数", "000001.SH"], ["深证成指", "399001.SZ"], ["创业板指", "399006.SZ"], ["中证100", "000903.SH"], ["中证100", "399903.SZ"]], columns=['name', 'value']) # 世界的指数 GLOBAL_INDEX = ['道琼斯', '标普500', '纳斯达克', '恒生指数', '日经225', '韩国KOSPI', '美元指数'] METAL_INDEX = pd.DataFrame( [["黄金", "Au99.99"] # , ["白银", "Ag99.99"] ], columns=['name', 'value']) def __init__(self, trade_date): self.trade_date = trade_date.strftime('%Y%m%d') self.start_date = return_trading_day(trade_date, -29) self.end_date = self.trade_date def calendar_page_json(self): """ 日历页面数据，只要A股指数获取1年内的全部数控，生成一个json文件 :return: """ # 计算一年前的日期 start_date = datetime.strptime(self.end_date, '%Y%m%d') - timedelta(days=365) start_date = start_date.strftime("%Y%m%d") # 在首页显示这三个的当日最新涨跌情况 index_page_display = ["000001.SH", "399001.SZ", "399006.SZ"] res_pandas = pandas.DataFrame(columns=['title', 'start', 'textColor', 'backgroundColor']) for stock in index_page_display: for _ in range(5): try: # 获取单个指数日线行情 df = xcsc_pro.index_daily( ts_code=stock, start_date=start_date, end_date=self.end_date, fields="ts_code,trade_date,pct_chg") break except Exception as e: print("获取单个指数日线行情出现问题：", e) time.sleep(1) continue else: print("重试5次依然没有解决问题，请重视~！！！") return None # 获取对应的中文名称 name = self.ZH_INDEX.loc[self.ZH_INDEX['value'] == stock, 'name'].iloc[0] df['pct_chg_format'] = (round(df['pct_chg'], 2)).astype("str") + "%" df['title'] = name + " --> " + df['pct_chg_format'] # 将字符串转换为日期时间对象 df['trade_date'] = pd.to_datetime(df['trade_date']) # 格式化日期时间列为 "YYYY-MM-DD" 形式 df['start'] = df['trade_date'].dt.strftime("%Y-%m-%d") # 设置条件 conditions = [ (df['pct_chg'] > 1), (df['pct_chg'] < -1), ((-1 <= df['pct_chg']) & (df['pct_chg'] < 0)), ((0 < df['pct_chg']) & (df['pct_chg'] <= 1)) ] # 字体颜色当涨跌幅大于1的时候，将字体颜色设置成白色 #e60000 text_colors = ['#ffffff', '#ffffff', '#5fcf57', '#e60000'] df['textColor'] = np.select(conditions, text_colors) # 背景颜色 background_colors = ['#e60000', '#5fcf57', '#ffffff', '#ffffff'] df['backgroundColor'] = np.select(conditions, background_colors) df = df[['title', 'start', 'textColor', 'backgroundColor']] res_pandas = pd.concat([res_pandas, df], ignore_index=True) return res_pandas def html_page_data(self): # A股指数 a_all_df, a_table_date, a_home_page_df = self.a_data() # 全球部分指数 g_all_df, g_table_date, g_home_page_df = self.global_index() # 贵金属行情 s_all_df, s_table_date, s_home_page_df = self.spot_hist_sge() # 生成首页的徽章 badge = "" badge += self.__create_badge(a_home_page_df) badge += self.__create_badge(g_home_page_df) badge += self.__create_badge(s_home_page_df) # 生成指数数据页 card_deck_df = pd.concat([a_table_date, g_table_date], axis=0) card_deck_df = pd.concat([card_deck_df, s_table_date], axis=0) card_deck = self.__create_card_deck(card_deck_df) return badge, card_deck def __create_badge(self, df): # 生成首页的徽章 badge = "" for index, row in df.iterrows(): badge_color = "badge-success" if row['pct_chg'] > 0: badge_color = "badge-danger" badge += f"

{row['name']}：{row['pct_chg']}

" return badge def __create_card_deck(self, df): grouped = df.groupby('name') card_deck = '

' row = 1 for group_name, group_df in grouped: card_deck_color = "border-success" if group_df.head(1)["pct_chg"].values[0] > 0: card_deck_color = "border-danger" card_deck += f'

' \ f'

{group_name}

' \ f'

0 else "text-success"}">' \ f'

{round(group_df.head(1)["close"].values[0], 2)}

' \ f'

' \ f' 0 else "text-success"} ml-3">' \ f'{round(group_df.head(2)["close"].values[1], 2)}' \ f'' \ f' 0 else "text-success"} ml-auto mr-3">' \ f'{round(group_df.head(3)["close"].values[2], 2)}' \ f'' \ f'

' \ f'

' if row % 5 == 0: card_deck += '

' print(f"{row - 1} -- >> {len(grouped)}") if row != len(grouped): card_deck += '

' if row % 5 != 0 and row == len(grouped): card_deck += '

' row += 1 return card_deck def a_data(self): """ A股指数数据返回 1个月内，3日，1日 :return: """ # 在首页显示这三个的当日最新涨跌情况 index_page_display = ["000001.SH", "399001.SZ", "399006.SZ"] all_df, home_page_df, table_df = pd.DataFrame(), pd.DataFrame(), pd.DataFrame() for index, row in self.ZH_INDEX.iterrows(): print(self.start_date, " --- >>", self.end_date) # 获取单个指数日线行情 for _ in range(5): try: df = xcsc_pro.index_daily(ts_code=row['value'], start_date=self.start_date, end_date=self.end_date, fields="ts_code,trade_date,pct_chg,close") break except Exception as e: print("获取单个指数日线行情出现问题：", e) continue else: print("重试5次依然没有解决问题，请重视~！！！") return None df['pct_chg'] = df['pct_chg'].round(2) df['name'] = self.ZH_INDEX.loc[self.ZH_INDEX['value'] == row['value'], 'name'].iloc[0] # 一个月的数据 all_df = pd.concat([all_df, df], axis=0) if index_page_display.count(row['value']): print(df) # 一日的数据 home_page_date = df.head(1) home_page_df = pd.concat([home_page_df, home_page_date], axis=0) # 三日的数据 table_df_date = df.head(3) table_df = pd.concat([table_df, table_df_date], axis=0) return all_df, table_df, home_page_df def global_index(self): """ 国外部分指数数据 3日内的涨跌幅 1个月内的折线图 :return: """ all_df, table_df, home_page_df = pd.DataFrame(), pd.DataFrame(), pd.DataFrame() for i in self.GLOBAL_INDEX: for _ in range(5): try: df = qs.get_data(i) break except Exception as e: print(f"国外部分指数数据【{i}】拉取出现错误，", e) continue else: print("重试5次依然没有解决问题，请重视~！！！") return None df['pct_chg'] = round(((df['close'] - df['open']) / df['open']) * 100, 2) all_df_date, table_df_date, home_page_data = df.tail(30), df.tail(3), df.tail(1) all_df = pd.concat([all_df, all_df_date], axis=0) table_df = pd.concat([table_df, table_df_date], axis=0) home_page_df = pd.concat([home_page_df, home_page_data], axis=0) # pct_chg = global_df.groupby('name')['pct_chg'].apply(list).reset_index() # print(pct_chg) return all_df, table_df, home_page_df def oil_index(self): # 油价 # 调价日期 energy_oil_hist_df = ak.energy_oil_hist() print(energy_oil_hist_df) formatted_date = energy_oil_hist_df['调整日期'].iloc[-3].strftime('%Y%m%d') # 地区油价 energy_oil_detail_df = ak.energy_oil_detail(date=formatted_date) energy_oil_detail_df = energy_oil_detail_df[energy_oil_detail_df['地区'] == '北京'] print(energy_oil_detail_df) def spot_hist_sge(self): all_df, table_df, home_page_df = pd.DataFrame(), pd.DataFrame(), pd.DataFrame() for index, row in self.METAL_INDEX.iterrows(): for _ in range(5): try: df = ak.spot_hist_sge(symbol=row['value']) break except Exception as e: print(f"上海黄金交易所-数据资讯-行情走势-历史数据，", e) continue else: print("重试5次依然没有解决问题，请重视~！！！") return None df['name'] = self.METAL_INDEX.loc[self.METAL_INDEX['value'] == row['value'], 'name'].iloc[0] df['pct_chg'] = round(((df['close'] - df['open']) / df['open']) * 100, 2) all_df_date, table_df_date, home_page_data = df.tail(30), df.tail(3), df.tail(1) all_df = pd.concat([all_df, all_df_date], axis=0) table_df = pd.concat([table_df, table_df_date], axis=0) home_page_df = pd.concat([home_page_df, home_page_data], axis=0) return all_df, table_df, home_page_df if __name__ == '__main__': # current_date = datetime.now().strftime('%Y%m%d') current_date = datetime.strptime('20230918', "%Y%m%d") # if if_run(current_date): index = AllIndex(trade_date=current_date) # index.a_index() badge, card_deck = index.html_page_data() print(badge)