This commit is contained in:
liy
2026-10-21 19:16:29 +08:00
parent f23935f331
commit dadd8a70b5
6 changed files with 304 additions and 0 deletions
+14
View File
@@ -0,0 +1,14 @@
{
"2026-03-27": {
"data_date": "2026-03-27 16:11:54+08:00",
"turnover_yi": 25.3,
"share_yi": 451.52,
"cached_at": "2026-03-30 00:30:15"
},
"2026-03-26": {
"data_date": "2026-03-26 16:11:54+08:00",
"turnover_yi": 25.3,
"share_yi": 451.52,
"cached_at": "2026-03-29 18:04:54"
}
}
+174
View File
@@ -0,0 +1,174 @@
# -*- coding: utf-8 -*-
import akshare as ak
from datetime import datetime, timedelta
import json
import os
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import time
# 缓存文件路径
CACHE_FILE = "./fund_cache.json"
# 缓存保留天数
CACHE_DAYS = 10
# 基金代码
FUND_CODE = "510300"
# 网络请求重试配置
RETRY_TIMES = 3 # 重试次数
RETRY_DELAY = 2 # 重试间隔(秒)
TIMEOUT = 10 # 请求超时时间(秒)
# ---------------------- 网络请求配置 ----------------------
def get_retry_session():
"""创建带重试机制的请求会话"""
session = requests.Session()
retry = Retry(
total=RETRY_TIMES,
backoff_factor=RETRY_DELAY,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET", "POST"]
)
adapter = HTTPAdapter(max_retries=retry)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
# ---------------------- 缓存管理函数 ----------------------
def load_cache():
"""加载缓存文件"""
if os.path.exists(CACHE_FILE):
with open(CACHE_FILE, "r", encoding="utf-8") as f:
return json.load(f)
return {}
def save_cache(cache_data):
"""保存缓存到文件"""
with open(CACHE_FILE, "w", encoding="utf-8") as f:
json.dump(cache_data, f, ensure_ascii=False, indent=2)
def clean_expired_cache(cache_data):
"""清理超过 10 天的缓存"""
cutoff_date = (datetime.now() - timedelta(days=CACHE_DAYS)).strftime("%Y-%m-%d")
expired_keys = [k for k in cache_data.keys() if k < cutoff_date]
for key in expired_keys:
del cache_data[key]
return cache_data
def get_cache_key(date_str):
"""生成缓存键(只保留日期部分)"""
return date_str[:10] if len(date_str) >= 10 else date_str
# ---------------------- 数据获取函数 ----------------------
def fetch_fund_data(fund_code):
"""从 AKShare 获取基金实时数据(增加重试和超时)"""
# 替换 akshare 的默认请求会话,增加重试
ak.session = get_retry_session()
for attempt in range(RETRY_TIMES):
try:
# 设置超时时间,避免无限等待
etf_spot_df = ak.fund_etf_spot_em(timeout=TIMEOUT)
fund_data = etf_spot_df[etf_spot_df["代码"] == fund_code].iloc[0]
# 获取数据日期
data_date = fund_data.get("更新时间", fund_data.get("日期", "未知"))
# 原始数据(单位:元、份)
turnover_yuan = fund_data["成交额"]
latest_share = fund_data["最新份额"]
# 换算为亿元、亿份
turnover_yi = turnover_yuan / 100000000
share_yi = latest_share / 100000000
return {
"data_date": str(data_date),
"turnover_yi": turnover_yi,
"share_yi": share_yi
}
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:
if attempt < RETRY_TIMES - 1:
print(f"⚠️ 网络请求失败(第 {attempt+1} 次),{RETRY_DELAY} 秒后重试... 错误:{e}")
time.sleep(RETRY_DELAY)
continue
raise # 最后一次失败则抛出异常
except Exception as e:
raise # 其他错误直接抛出
def get_cached_or_fetch(fund_code, cache_data):
"""获取缓存数据或从网络获取"""
today_key = get_cache_key(datetime.now().strftime("%Y-%m-%d"))
if today_key in cache_data:
print(f"📂 使用缓存数据:{today_key}")
return cache_data[today_key], True
print(f"🌐 获取网络数据...")
data = fetch_fund_data(fund_code)
today_key = get_cache_key(data["data_date"])
# 保存到缓存(保留两位小数)
cache_data[today_key] = {
"data_date": data["data_date"],
"turnover_yi": round(data["turnover_yi"], 2),
"share_yi": round(data["share_yi"], 2),
"cached_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}
save_cache(cache_data)
print(f"💾 已保存缓存:{today_key}")
return cache_data[today_key], False
# ---------------------- 数据显示函数 ----------------------
def print_fund_info(fund_code, data):
"""打印基金信息"""
print("=" * 60)
print(f"📊 基金:{fund_code} | 查询时间:{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print("=" * 60)
print(f"📅 数据日期:{data['data_date']}")
print(f"✅ 当日成交额:{data['turnover_yi']:,.2f} 亿元")
print(f"✅ 最新总份额:{data['share_yi']:,.2f} 亿份")
print("=" * 60)
# ---------------------- 主函数 ----------------------
def get_etf_data():
"""主程序入口"""
# 加载并清理缓存
cache_data = load_cache()
cache_data = clean_expired_cache(cache_data)
try:
# 获取数据(缓存或网络)
data, is_cached = get_cached_or_fetch(FUND_CODE, cache_data)
# 打印结果
print_fund_info(FUND_CODE, data)
except Exception as e:
print(f"❌ 数据获取失败:{e}")
# 如果有缓存,尝试使用最近的缓存数据
if cache_data:
latest_date = max(cache_data.keys())
latest_data = cache_data[latest_date]
print(f"\n🔧 尝试使用最近缓存({latest_date})的数据:")
print_fund_info(FUND_CODE, latest_data)
finally:
# 清理过期缓存并保存
cache_data = clean_expired_cache(cache_data)
save_cache(cache_data)
# ---------------------- 程序入口 ----------------------
if __name__ == "__main__":
get_etf_data()
+33
View File
@@ -0,0 +1,33 @@
# -*- coding: utf-8 -*-
import requests
def push_gotify(title, message, priority=0):
# ========== 在这里改成你自己的信息 ==========
GOTIFY_URL = "http://1.14.127.179:40266" # 你的 Gotify 地址
APP_TOKEN = "AxoAjSFEhoFY_ln" # 你的 APP Token
url = f"{GOTIFY_URL}/message"
params = {
"token": APP_TOKEN,
"title": title,
"message": message,
"priority": priority
}
try:
resp = requests.post(url, params=params, timeout=10)
if resp.status_code in (200, 201):
print("推送成功")
else:
print(f"推送失败 {resp.status_code}: {resp.text}")
except Exception as e:
print(f"请求异常: {e}")
# 测试发送
if __name__ == "__main__":
push_gotify(
title="akshare 数据处理完成",
message="近10天数据已成功获取并处理,表格已生成",
priority=0
)
+4
View File
@@ -0,0 +1,4 @@
akshare==1.18.49
requests==2.33.0
pandas==3.0.1
numpy==2.4.3
+79
View File
@@ -0,0 +1,79 @@
# -*- coding: utf-8 -*-
import json
import os
from datetime import datetime
# 导入数据获取模块的函数
from get_etf_data import load_cache, CACHE_FILE, CACHE_DAYS, FUND_CODE,get_etf_data
# 导入推送模块的函数
from gotify import push_gotify
def format_cache_message(cache_data):
"""格式化缓存数据为消息内容"""
if not cache_data:
return "⚠️ 暂无缓存数据"
# 按日期排序(最近的数据在前)
sorted_dates = sorted(cache_data.keys(), reverse=True)
message_lines = []
for date in sorted_dates:
data = cache_data[date]
line = (
f"📅 {data['data_date'][:10]}\n"
f" 成交额:{data['turnover_yi']:,.2f} 亿元\n"
f" 总份额:{data['share_yi']:,.2f} 亿份"
)
message_lines.append(line)
return "\n\n".join(message_lines)
def send_cache_notification():
"""发送缓存数据通知"""
get_etf_data()
# 加载缓存
cache_data = load_cache()
if not cache_data:
push_gotify(
title="⚠️ ETF 数据通知",
message="暂无缓存数据,请先运行 get_etf_data.py 获取数据",
priority=5
)
return
# 统计数据
total_days = len(cache_data)
latest_date = max(cache_data.keys())
latest_data = cache_data[latest_date]
# 生成标题
title = f"📊 ETF 数据报告 ({FUND_CODE})"
# 生成消息内容
message = (
f"📈 缓存天数:{total_days} 天\n"
f"🕐 最新数据:{latest_date}\n"
f"💰 成交额:{latest_data['turnover_yi']:,.2f} 亿元\n"
f"📦 总份额:{latest_data['share_yi']:,.2f} 亿份\n\n"
f"{'='*10}\n"
f"{format_cache_message(cache_data)}"
)
# 发送推送
push_gotify(
title=title,
message=message,
priority=3
)
print(f"✅ 已发送 {total_days} 天的缓存数据通知")
# ---------------------- 程序入口 ----------------------
if __name__ == "__main__":
send_cache_notification()
View File