fish
202 lines
7.6 KiB
Python
202 lines
7.6 KiB
Python
"""拉取建材价格指数，更新 data.json 并刷新 index.html。"""
|
|
import json
|
|
import pandas as pd
|
|
import akshare as ak
|
|
|
|
df = ak.macro_china_construction_index()
|
|
df['日期'] = pd.to_datetime(df['日期'])
|
|
df = df.sort_values('日期').reset_index(drop=True)
|
|
|
|
for w in [5, 10, 20, 60]:
|
|
df[f'MA{w}'] = df['最新值'].rolling(w).mean().round(1)
|
|
df[f'MA{w}_dir'] = df[f'MA{w}'].diff().apply(lambda x: 'up' if x > 0 else ('down' if x < 0 else 'flat'))
|
|
|
|
latest = df.iloc[-1]
|
|
|
|
signals = []
|
|
if latest['最新值'] > latest['MA5']:
|
|
signals.append(('短期', '偏多', '价格 > MA5'))
|
|
else:
|
|
signals.append(('短期', '偏空', '价格 < MA5'))
|
|
if latest['MA5'] > latest['MA20']:
|
|
signals.append(('中期', '偏多', 'MA5 > MA20'))
|
|
else:
|
|
signals.append(('中期', '偏空', 'MA5 < MA20'))
|
|
if latest['MA20'] > latest['MA60']:
|
|
signals.append(('长期', '偏多', 'MA20 > MA60'))
|
|
else:
|
|
signals.append(('长期', '偏空', 'MA20 < MA60'))
|
|
|
|
momentum_5d = latest['最新值'] - df.iloc[-6]['最新值']
|
|
signals.append(('动量(5日)', '偏多' if momentum_5d > 0 else '偏空', f"{momentum_5d:+d}"))
|
|
|
|
bull_count = sum(1 for _, s, _ in signals if s == '偏多')
|
|
if bull_count >= 3:
|
|
overall = ('偏多', '#16a34a')
|
|
elif bull_count <= 1:
|
|
overall = ('偏空', '#dc2626')
|
|
else:
|
|
overall = ('震荡', '#d97706')
|
|
|
|
summary = {
|
|
'date': latest['日期'].strftime('%Y-%m-%d'),
|
|
'value': int(latest['最新值']),
|
|
'change': round(float(latest['涨跌幅']), 4),
|
|
'chg_3m': round(float(latest['近3月涨跌幅']), 4),
|
|
'chg_6m': round(float(latest['近6月涨跌幅']), 4),
|
|
'chg_1y': round(float(latest['近1年涨跌幅']), 4),
|
|
}
|
|
|
|
mas = {}
|
|
for w in [5, 10, 20, 60]:
|
|
mas[f'MA{w}'] = {
|
|
'value': float(latest[f'MA{w}']),
|
|
'direction': latest[f'MA{w}_dir'],
|
|
'price_vs_ma': 'above' if latest['最新值'] > latest[f'MA{w}'] else 'below'
|
|
}
|
|
|
|
rows = []
|
|
for _, r in df.tail(30).iterrows():
|
|
rows.append({
|
|
'date': r['日期'].strftime('%m-%d'),
|
|
'value': int(r['最新值']),
|
|
'chg': round(float(r['涨跌幅']), 4),
|
|
'MA5': None if pd.isna(r['MA5']) else round(float(r['MA5']), 1),
|
|
'MA20': None if pd.isna(r['MA20']) else round(float(r['MA20']), 1),
|
|
'MA60': None if pd.isna(r['MA60']) else round(float(r['MA60']), 1),
|
|
})
|
|
|
|
|
|
def build_history(df):
|
|
"""构建历史全景数据：月度采样 + 周期标注 + 统计。"""
|
|
|
|
# 月度采样（取每月最后一条有效记录）
|
|
monthly = df.set_index('日期').resample('ME')['最新值'].last().dropna().reset_index()
|
|
ma60_monthly = df.set_index('日期').resample('ME')['MA60'].last().dropna().reset_index()
|
|
monthly_vals = []
|
|
for _, r in monthly.iterrows():
|
|
ma_row = ma60_monthly[ma60_monthly['日期'] == r['日期']]
|
|
ma60_val = None if ma_row.empty or pd.isna(ma_row['MA60'].iloc[0]) else round(float(ma_row['MA60'].iloc[0]), 1)
|
|
monthly_vals.append({
|
|
'date': r['日期'].strftime('%Y-%m'),
|
|
'value': int(r['最新值']),
|
|
'ma60': ma60_val,
|
|
})
|
|
|
|
# 历史统计
|
|
vals = df['最新值']
|
|
latest_val = float(vals.iloc[-1])
|
|
pct_rank = round((vals < latest_val).sum() / len(vals) * 100, 1)
|
|
|
|
# 年度涨跌（排除不足 200 个交易日的首尾年）
|
|
df['year'] = df['日期'].dt.year
|
|
yearly = df.groupby('year')['最新值'].agg(['first', 'last', 'count'])
|
|
yearly = yearly[yearly['count'] >= 200]
|
|
yearly['chg'] = ((yearly['last'] - yearly['first']) / yearly['first'] * 100).round(1)
|
|
down_years = int((yearly['chg'] < 0).sum())
|
|
ytd = float(yearly['chg'].iloc[-1]) if len(yearly) > 0 else 0
|
|
|
|
# 识别主要周期：用未取整的 MA60 找极值，间距至少 4 个月
|
|
ma60_raw = df['最新值'].rolling(60).mean().values # 未取整，保留精度
|
|
all_turns = [] # (date, value, type)
|
|
min_gap = 80 # 交易日，约 4 个月
|
|
last_turn_idx = -999
|
|
for i in range(2, len(ma60_raw) - 2):
|
|
if pd.isna(ma60_raw[i]):
|
|
continue
|
|
if ma60_raw[i] > ma60_raw[i - 1] and ma60_raw[i] > ma60_raw[i - 2] and ma60_raw[i] > ma60_raw[i + 1] and ma60_raw[i] > ma60_raw[i + 2]:
|
|
if i - last_turn_idx >= min_gap:
|
|
all_turns.append((df['日期'].iloc[i], float(ma60_raw[i]), 'peak'))
|
|
last_turn_idx = i
|
|
elif ma60_raw[i] < ma60_raw[i - 1] and ma60_raw[i] < ma60_raw[i - 2] and ma60_raw[i] < ma60_raw[i + 1] and ma60_raw[i] < ma60_raw[i + 2]:
|
|
if i - last_turn_idx >= min_gap:
|
|
all_turns.append((df['日期'].iloc[i], float(ma60_raw[i]), 'trough'))
|
|
last_turn_idx = i
|
|
|
|
# 构建周期（谷 → 下一个峰）
|
|
cycles = []
|
|
for i, (t_date, t_val, t_type) in enumerate(all_turns):
|
|
if t_type != 'trough':
|
|
continue
|
|
# 找下一个峰
|
|
for j in range(i + 1, len(all_turns)):
|
|
if all_turns[j][2] == 'peak':
|
|
p_date, p_val, _ = all_turns[j]
|
|
gain = round((p_val - t_val) / t_val * 100, 1)
|
|
if gain > 10:
|
|
label = ''
|
|
if t_date.year == 2015 or t_date.year == 2016:
|
|
label = '供给侧改革'
|
|
elif t_date.year == 2020:
|
|
label = '疫情后刺激'
|
|
cycles.append({
|
|
'label': label,
|
|
'start': t_date.strftime('%Y-%m'),
|
|
'end': p_date.strftime('%Y-%m'),
|
|
'from': int(t_val),
|
|
'to': int(p_val),
|
|
'pct': gain,
|
|
})
|
|
break
|
|
|
|
# 当前反弹（用价格实际谷底，不是 MA60 谷底）
|
|
# 找最近 2 年的价格最低点
|
|
recent_2y = df[df['日期'] >= df['日期'].iloc[-1] - pd.DateOffset(years=2)]
|
|
trough_idx = recent_2y['最新值'].idxmin()
|
|
trough_date = df.loc[trough_idx, '日期']
|
|
trough_val = float(df.loc[trough_idx, '最新值'])
|
|
cur_gain = round((latest_val - trough_val) / trough_val * 100, 1)
|
|
current_rebound = {
|
|
'label': '当前反弹',
|
|
'start': trough_date.strftime('%Y-%m'),
|
|
'end': None,
|
|
'from': int(trough_val),
|
|
'to': int(latest_val),
|
|
'pct': cur_gain,
|
|
}
|
|
|
|
# 均线结构
|
|
last = df.iloc[-1]
|
|
ma20_vs_ma60 = 'above' if last['MA20'] > last['MA60'] else 'below'
|
|
|
|
return {
|
|
'monthly': monthly_vals,
|
|
'stats': {
|
|
'mean': round(float(vals.mean()), 1),
|
|
'min': int(vals.min()),
|
|
'max': int(vals.max()),
|
|
'min_date': df.loc[vals.idxmin(), '日期'].strftime('%Y-%m'),
|
|
'max_date': df.loc[vals.idxmax(), '日期'].strftime('%Y-%m'),
|
|
'percentile': pct_rank,
|
|
'down_years': down_years,
|
|
'ytd_chg': round(ytd, 1),
|
|
},
|
|
'cycles': cycles,
|
|
'current_rebound': current_rebound,
|
|
'ma20_vs_ma60': ma20_vs_ma60,
|
|
}
|
|
|
|
|
|
output = {
|
|
'summary': summary,
|
|
'mas': mas,
|
|
'signals': [{'period': p, 'signal': s, 'detail': d} for p, s, d in signals],
|
|
'overall': {'signal': overall[0], 'color': overall[1]},
|
|
'recent': rows,
|
|
'history': build_history(df),
|
|
}
|
|
|
|
with open('data.json', 'w') as f:
|
|
json.dump(output, f, ensure_ascii=False, indent=2)
|
|
|
|
SENTINEL = '/*__DATA_SENTINEL__*/{}'
|
|
html = open('index_template.html').read()
|
|
html = html.replace(SENTINEL, json.dumps(output, ensure_ascii=False))
|
|
open('index.html', 'w').write(html)
|
|
|
|
print(f"刷新完成: {summary['date']} → {summary['value']} (日涨跌 {summary['change']:+.2f}%)")
|
|
print(f"综合信号: {overall[0]}")
|
|
print("信号明细:")
|
|
for p, s, d in signals:
|
|
print(f" {p}: {s} ({d})")
|