from dash import html
from dash import dcc
from dash import Input#输入
from dash import Output#输出
import time
from dash import dash_table
import dash
import pandas as pd
import requests
import json
from datetime import datetime
import os
path=os.path.dirname(os.path.abspath(__file__))
now_date=str(datetime.now())[:10]
select_list=['用户信息','实时数据','历史数据','清空实时数据','清空历史数据']
def params_func(text=''):
'''
解析函数
'''
data_list=[]
f=text.split('\n')
for i in f:
text=i.strip().lstrip()
data_list.append(text)
func='\n'.join(data_list)
return func
def get_n1_n2_daily(start_date='2023-12-14',end_date='2023-12-30'):
'''
获取2个时间的天数
'''
import datetime
end_date=end_date.split('-')
start_date=start_date.split('-')
d1 =datetime.datetime(int(start_date[0]),int(start_date[1]),int(start_date[2]))
d2 = datetime.datetime(int(end_date[0]),int(end_date[1]),int(end_date[2])) # 第二个日期
interval = d2 - d1 # 两日期差距
days=interval.days
return days
app=dash.Dash(__name__)
app.layout=html.Div([
html.H3('小果聚合跟单服务器,找作者获取授权码 微信15117320079,输入的数据必须是字符串'),
html.P('实盘前请清空实时数据其他的不用,点击运行按钮获取数据,运行实盘前需要获取授权码'),
html.Tbody([
html.Tr([
html.Td('用户名称',
style={'border':'1px solid','width':'200px'}),
html.Td('数据类型',
style={'border':'1px solid','width':'200px'}),
html.Td('运行程序',
style={'border':'1px solid','width':'200px'}),
html.Td('下载数据',
style={'border':'1px solid','width':'200px'})
]),
html.Tr([
html.Td(dcc.Input(value='123456',id='joinquant_trader_password'),
style={'border':'1px solid','width':'200px'}),
html.Td(dcc.Dropdown(options=select_list,value='用户信息',id='joinquant_trader_data_type'),
style={'border':'1px solid','width':'200px'}),
html.Td(dcc.Dropdown(options=['运行','不运行'],id='joinquant_trader_run',value='不运行'),
style={'border':'1px solid','width':'200px'}),
html.Td(dcc.RadioItems(options={'下载数据':"下载数据","不下载数据":"不下载数据"},id='joinquant_trader_down_data',value='不下载数据'),
style={'border':'1px solid','width':'200px'})
]),
]),
dcc.Textarea(value="""
{'状态': 'held', '订单添加时间': 'datetime.datetime(2024, 4, 23, 9, 30)', '买卖': 'False', '下单数量': '9400', '已经成交': '9400', '股票代码': '', '订单ID': '1732208241', '平均成交价格': '10.5', '持仓成本': '10.59', '多空': 'long', '交易费用': '128.31'}
""",
id='joinquant_trader_text',style={'width': '80%', 'height': 200,"text-align":"left",}),
html.H3('数据展示'),
dash_table.DataTable(
id='joinquant_trader_table',
page_size=10,
style_table={'font-size': 15},
sort_action='native'),
dcc.Download(id='joinquant_trader_table_down')
])
#回调表1
@app.callback(
Output(component_id='joinquant_trader_table',component_property='data',allow_duplicate=True),
Input(component_id='joinquant_trader_password',component_property='value'),
Input(component_id='joinquant_trader_data_type',component_property='value'),
Input(component_id='joinquant_trader_text',component_property='value'),
Input(component_id='joinquant_trader_run',component_property='value'),
Input(component_id='joinquant_trader_down_data',component_property='value'),
prevent_initial_call=True
)
def show_wencai_data_table_1(password,data_type,text,run,down_data):
if run=='运行':
password_df=pd.read_excel(r'授权表\授权表.xlsx',dtype='object')
try:
del password_df['Unnamed: 0']
except:
pass
now_date=str(datetime.now())[:10]
password_df['用户名称']=password_df['用户名称'].astype(str)
password_df['到期时间']=password_df['到期时间'].astype(str)
password_df.index=password_df['用户名称']
password_list=password_df['用户名称'].tolist()
if str(password) in password_list:
text=params_func(text)
text=eval(text)
df=password_df[password_df['用户名称']==password]
end_date=str(df['到期时间'].tolist()[-1])[:10]
day=get_n1_n2_daily(start_date=now_date,end_date=end_date)
df['到期天数']=day
if day>=1:
if data_type=='用户信息':
df['信息推送']='授权码正常'
df['数据状态']=True
elif data_type=='实时数据':
try:
df=pd.read_csv(r'{}\data\实时数据\{}实时数据.csv'.format(path,password))
try:
del df['Unnamed: 0']
except:
pass
df['交易日']=df['交易日'].astype(str)
select_list=list(text.keys())
select_list.append('交易日')
select_list.append('数据状态')
df=df[select_list]
df=df[df['交易日']==now_date]
except Exception as e:
print(e,'实时数据有问题')
df=pd.DataFrame()
try:
hist=pd.read_csv(r'{}\data\历史数据\{}历史数据.csv'.format(path,password))
try:
del hist['Unnamed: 0']
except:
pass
df['交易日']=df['交易日'].astype(str)
select_list=list(text.keys())
select_list.append('交易日')
select_list.append('数据状态')
df=df[select_list]
except Exception as e:
print(e,'历史数据有问题')
hist=pd.DataFrame()
data=pd.DataFrame()
data['状态']=[text['状态']]
data['订单添加时间']=[text['订单添加时间']]
data['买卖']=[text['买卖']]
data['下单数量']=[text['下单数量']]
data['已经成交']=[text['已经成交']]
data['股票代码']=[text['股票代码']]
data['订单ID']=[text['订单ID']]
data['平均成交价格']=[text['平均成交价格']]
data['持仓成本']=[text['持仓成本']]
data['多空']=[text['多空']]
data['交易费用']=[text['交易费用']]
df=pd.concat([df,data],ignore_index=True)
df['证券代码']=df['股票代码'].apply(lambda x: str(x).split('.XSHE')[0])
df['数据长度']=df['证券代码'].apply(lambda x: len(str(x)))
df=df[df['数据长度']>=6]
try:
del df['数据长度']
except:
pass
df=df.drop_duplicates(subset=['股票代码','下单数量','买卖','多空'],keep='last')
df['交易日']=now_date
df['数据状态']=True
df.to_csv(r'{}\data\实时数据\{}实时数据.csv'.format(path,password))
hist=pd.concat([hist,data],ignore_index=True)
hist=hist.drop_duplicates(subset=['股票代码','下单数量','买卖','多空'],keep='last')
hist['交易日']=now_date
hist['数据状态']=True
hist.to_csv(r'{}\data\历史数据\{}历史数据.csv'.format(path,password))
elif data_type=='清空实时数据':
df=pd.DataFrame()
df.to_csv(r'{}\data\实时数据\{}实时数据.csv'.format(path,password))
elif data_type=='历史数据':
try:
hist=pd.read_csv(r'{}\data\历史数据\{}历史数据.csv'.format(path,password))
hist['证券代码']=hist['股票代码'].apply(lambda x: str(x).split('.XSHE')[0])
hist['数据长度']=hist['证券代码'].apply(lambda x: len(str(x)))
hist=hist[hist['数据长度']>=6]
hist=hist.drop_duplicates(subset=['股票代码','下单数量','买卖','多空'],keep='last')
try:
del hist['数据长度']
except:
pass
try:
del hist['Unnamed: 0']
except:
pass
df=hist
df['交易日']=df['交易日'].astype(str)
select_list=list(text.keys())
select_list.append('交易日')
select_list.append('数据状态')
df=df[select_list]
except Exception as e:
print(e,'历史数据有问题')
df=pd.DataFrame()
elif data_type=='清空历史数据':
df=pd.DataFrame()
df.to_csv(r'{}\data\历史数据\{}历史数据.csv'.format(path,password))
else:
df['信息推送']='授权码到期找作者获取'
df['数据状态']=False
else:
df['信息推送']='授权码到期找作者获取'
df['数据状态']=False
return df.to_dict('records')
else:
df=pd.DataFrame()
df['信息提示']=['用户没有注册请注册/联系作者微信15117320079']
df['数据状态']=False
return df.to_dict('records')
else:
df=pd.DataFrame()
df['信息提示']=['没有点击运行选择']
df['数据状态']=False
return df.to_dict('records')
#回调下载
@app.callback(
Output(component_id='joinquant_trader_table_down',component_property='data',allow_duplicate=True),
Input(component_id='joinquant_trader_password',component_property='value'),
Input(component_id='joinquant_trader_data_type',component_property='value'),
Input(component_id='joinquant_trader_text',component_property='value'),
Input(component_id='joinquant_trader_run',component_property='value'),
Input(component_id='joinquant_trader_down_data',component_property='value'),
prevent_initial_call=True
)
def show_wencai_data_table_down(password,data_type,text,run,down_data):
if down_data=='下载数据':
password_df=pd.read_excel(r'授权表\授权表.xlsx',dtype='object')
try:
del password_df['Unnamed: 0']
except:
pass
now_date=str(datetime.now())[:10]
password_df['用户名称']=password_df['用户名称'].astype(str)
password_df['到期时间']=password_df['到期时间'].astype(str)
password_df.index=password_df['用户名称']
password_list=password_df['用户名称'].tolist()
if str(password) in password_list:
text=params_func(text)
text=eval(text)
df=password_df[password_df['用户名称']==password]
end_date=str(df['到期时间'].tolist()[-1])[:10]
day=get_n1_n2_daily(start_date=now_date,end_date=end_date)
df['到期天数']=day
if day>=1:
if data_type in ['实时数据','历史数据']:
hist=pd.read_csv(r'{}\data\{}\{}{}.csv'.format(path,data_type,password,data_type))
try:
del hist['Unnamed: 0']
except:
pass
df=hist
return dcc.send_data_frame(df.to_excel,filename='{}{}.xlsx'.format(now_date,data_type))
if __name__=='__main__':
app.run(
host='0.0.0.0', # 监听所有网络接口
port=8888, # 端口
debug=True # 开发时建议开启debug模式，便于调试
)