251 lines
9.1 KiB
Python
251 lines
9.1 KiB
Python
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"<div class='col-md-auto'><span class='badge badge-pill {badge_color}'>{row['name']}：{row['pct_chg']}</span></div>"
|
||
return badge
|
||
|
||
def __create_card_deck(self, df):
|
||
grouped = df.groupby('name')
|
||
card_deck = '<div class="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'<div class="card {card_deck_color} mb-3" style="max-width: 14rem;">' \
|
||
f'<div class="card-header">{group_name}</div>' \
|
||
f'<div class="card-body {"text-danger" if group_df.head(1)["pct_chg"].values[0] > 0 else "text-success"}">' \
|
||
f'<h4 class="card-title">{round(group_df.head(1)["close"].values[0], 2)}</h4>' \
|
||
f'<p class="card-text row">' \
|
||
f'<span class="{"text-danger" if group_df.head(2)["pct_chg"].values[1] > 0 else "text-success"} ml-3">' \
|
||
f'{round(group_df.head(2)["close"].values[1], 2)}' \
|
||
f'</span>' \
|
||
f'<span class="{"text-danger" if group_df.head(3)["pct_chg"].values[2] > 0 else "text-success"} ml-auto mr-3">' \
|
||
f'{round(group_df.head(3)["close"].values[2], 2)}' \
|
||
f'</span>' \
|
||
f'</p>' \
|
||
f'</div>' \
|
||
f'</div>'
|
||
if row % 5 == 0:
|
||
card_deck += '</div>'
|
||
print(f"{row - 1} -- >> {len(grouped)}")
|
||
if row != len(grouped):
|
||
card_deck += '<div class="card-deck">'
|
||
if row % 5 != 0 and row == len(grouped):
|
||
card_deck += '</div>'
|
||
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)
|