from
__future__
import
(absolute_import, division, print_function, unicode_literals)
import
time
import
akshare as ak
import
pandas as pd
import
plotly.express as px
from
dash
import
Dash, html, dcc,
Input
, Output
app
=
Dash(__name__)
def
create_figure():
try
:
df
=
ak.stock_sector_fund_flow_rank(
indicator
=
"今日"
,
sector_type
=
"行业资金流"
)
except
Exception as e:
print
(f
"数据获取失败:{e}"
)
return
df.columns
=
df.columns.
str
.replace(
"今日"
, '', regex
=
False
)
df
=
df.rename(columns
=
{
'名称'
:
'板块名称'
})
df[
'资金净流入(亿)'
]
=
round
(df[
'主力净流入-净额'
]
/
100000000
,
2
)
df
=
df.sort_values(
'主力净流入-净占比'
, ascending
=
False
).dropna()
min_negative
=
df[
'主力净流入-净占比'
].
min
()
df[
'size_weight'
]
=
df[
'主力净流入-净占比'
].
apply
(
lambda
x: (
1
/
abs
(min_negative)
*
(x
-
min_negative)
+
1
)
)
fig
=
px.treemap(
df,
path
=
[
'板块名称'
],
values
=
'size_weight'
,
color
=
'主力净流入-净占比'
,
color_continuous_scale
=
'RdYlGn_r'
,
color_continuous_midpoint
=
0
,
range_color
=
[df[
'主力净流入-净占比'
].
min
(), df[
'主力净流入-净占比'
].
max
()],
height
=
800
,
width
=
1600
,
title
=
'🎨行业板块主力资金净占比分布热力图 (📌数据更新于: {})'
.
format
(pd.Timestamp.now().strftime(
"%H:%M:%S"
)),
branchvalues
=
'total'
,
hover_data
=
{
'涨跌幅'
:
':.2f%'
,
'资金净流入(亿)'
:
':.2f'
,
'主力净流入-净占比'
:
':.2f%'
}
)
fig.update_traces(
texttemplate
=
(
"<b>%{label}</b><br>"
"📈%{customdata[0]:.2f}%<br>"
"💰%{customdata[1]:.2f}亿"
),
hovertemplate
=
(
"<b>%{label}</b><br>"
"📈涨跌幅: %{customdata[0]:.2f}%<br>"
"💰资金净流入: <b>%{customdata[1]:.2f}</b>亿<br>"
"⚖️主力净占比: %{customdata[2]:.2f}%"
),
textfont
=
dict
(size
=
12
, color
=
'black'
, family
=
"SimHei"
)
)
return
fig
app.layout
=
html.Div([
dcc.Graph(
id
=
'live-graph'
),
dcc.Interval(
id
=
'interval-component'
,
interval
=
5
*
1000
,
n_intervals
=
0
)
])
@app
.callback(
Output(
'live-graph'
,
'figure'
),
Input
(
'interval-component'
,
'n_intervals'
)
)
def
update_graph(n):
try
:
return
create_figure()
except
Exception as e:
print
(f
"更新失败: {str(e)}"
)
return
px.scatter(title
=
"⚠️数据加载失败,请检查网络连接"
)
if
__name__
=
=
"__main__"
:
app.run_server(debug
=
True
)