python代码实时获取基金、股票涨跌幅发送微信
关注公众号
#!/usr/bin/python3.6
import requests
import datetime
import json
import time
import re
from random import uniform
# 更新后的基金列表
TARGET_FUNDS = [
{"code": "008143", "name": "工银黄金ETF联接C", "is_qdii": False, "is_stock": False},
{"code": "004814", "name": "中欧红利优享灵活配置混合A", "is_qdii": False, "is_stock": False},
{"code": "010372", "name": "大成成长进取混合C", "is_qdii": False, "is_stock": False},
{"code": "025491", "name": "平安中证卫星产业指数C", "is_qdii": False, "is_stock": False},
{"code": "009994", "name": "嘉实创新先锋混合A", "is_qdii": False, "is_stock": False},
{"code": "023638", "name": "国泰A股电网设备ETF联接A", "is_qdii": False, "is_stock": False}
#{"code": "601899", "name": "紫金矿业", "is_qdii": False, "is_stock": True},
#{"code": "000878", "name": "云南铜业", "is_qdii": False, "is_stock": True}
]
RETRY_MAX = 4
RETRY_DELAY = (1, 2)
TIMEOUT = 15
USER_AGENTS = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
]
def get_random_header(fund_code=None, is_stock=False):
header = {
"User-Agent": USER_AGENTS[0],
"Accept": "*/*",
"Connection": "keep-alive",
}
if is_stock:
header["Referer"] = "https://quote.eastmoney.com/"
else:
header["Referer"] = "https://fund.eastmoney.com/"
return header
def retry_request(func):
def wrapper(*args, **kwargs):
for retry in range(RETRY_MAX):
try:
time.sleep(uniform(*RETRY_DELAY))
result = func(*args, **kwargs)
if result: return result
except Exception as e:
print(f"[{func.__name__}] 第{retry + 1}次重试：{str(e)}")
return None
return wrapper
@retry_request
def get_stock_data(stock_code, stock_name):
"""获取股票实时数据 - 优化版"""
# 优先使用东方财富 API，因为数据字段更全且稳定
try:
# 判断市场: 6开头沪市(sh/1)，0或3开头深市(sz/0)
market_id = "1" if stock_code.startswith("6") else "0"
secid = f"{market_id}.{stock_code}"
url = "https://push2.eastmoney.com/api/qt/stock/get"
params = {
"secid": secid,
"fields": "f43,f57,f58,f60,f168,f169,f170,f46,f44,f45,f47,f48,f3", # f3 是涨跌幅
"ut": "bd1d9ddb04089700cf9c27f6f7426281",
"invt": "2",
"fltt": "2",
}
response = requests.get(url, params=params, headers=get_random_header(is_stock=True), timeout=TIMEOUT)
json_data = response.json()
if json_data and json_data.get("data"):
data = json_data["data"]
# 如果 f43 (现价) 为 "-", 说明还没开盘或停牌
current_price = data.get("f43")
if current_price == "-": return None
return {
"fund_name": data.get("f58", stock_name),
"fund_code": stock_code,
"est_net": round(float(current_price), 2),
"est_change": round(float(data.get("f3", 0)), 2), # 直接使用官方涨跌幅字段
"yest_net": round(float(data.get("f60", 0)), 2),
"yest_date": datetime.datetime.now().strftime("%Y-%m-%d"),
"est_time": datetime.datetime.now().strftime("%H:%M:%S"),
"source": "东方财富(股票)",
"is_stock": True,
"high": data.get("f44"),
"low": data.get("f45"),
"open": data.get("f46"),
"volume": data.get("f47"),
"amount": data.get("f48")
}
except Exception as e:
print(f"股票数据解析失败: {e}")
return None
@retry_request
def get_regular_fund_data(fund_code, fund_name):
"""获取普通基金的实时估算数据"""
try:
url = f"https://fundgz.1234567.com.cn/js/{fund_code}.js?rt={int(time.time() * 1000)}"
response = requests.get(url, headers=get_random_header(), timeout=TIMEOUT)
if "jsonpgz" in response.text:
content = re.search(r"jsonpgz\((.*)\);", response.text).group(1)
json_data = json.loads(content)
return {
"fund_name": fund_name,
"fund_code": fund_code,
"est_net": round(float(json_data["gsz"]), 4),
"est_change": round(float(json_data["gszzl"]), 2),
"yest_net": round(float(json_data["dwjz"]), 4),
"yest_date": json_data["jzrq"],
"est_time": json_data["gztime"],
"source": "天天基金"
}
except Exception as e:
print(f"基金数据获取失败: {e}")
return None
# 这里保留你原来的 get_qdii_fund_data, format_fund_report, send_wechat_notification 等函数...
# [此处为了篇幅省略，逻辑与你原代码一致，只需确保调用即可]
def format_fund_report(fund_data, fund_name, fund_code, is_qdii, is_stock):
"""格式化报告逻辑不变"""
if not fund_data:
return f"❌ {fund_name}（{fund_code}）：无法获取数据"
change_sign = "📈" if fund_data["est_change"] > 0 else "📉" if fund_data["est_change"] < 0 else "📊"
report = [f"📌 {fund_data['fund_name']}（{fund_data['fund_code']}）", ""]
if is_stock:
report.extend([
f"🏦 类型：股票",
f"⏰ 数据来源（{fund_data['source']}）",
f"💰 当前价格：¥{fund_data['est_net']}",
f"{change_sign} 涨跌幅：{fund_data['est_change']}%",
f"📅 昨收价格：¥{fund_data['yest_net']}",
f"🕐 更新时间：{fund_data['est_time']}"
])
else:
report.extend([
f"💴 净值/估值：¥{fund_data['est_net']}",
f"{change_sign} 涨跌幅：{fund_data['est_change']}%",
f"🕒 数据时间：{fund_data['est_time']}"
])
if not is_qdii:
report.append(f"📅 昨日净值：¥{fund_data['yest_net']}（{fund_data['yest_date']}）")
return "\n".join(report)
# ... 保持 main() 逻辑不变 ...
def send_wechat_notification(title, content):
url = "你的url信息"
data = {"text": title, "desp": content}
try:
requests.post(url, data=data, timeout=15)
return "✅ 发送成功"
except:
return "❌ 发送失败"
def main():
current_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
all_report = [f"# 监控报告（{current_time}）", ""]
for fund in TARGET_FUNDS:
f_code, f_name = fund["code"], fund["name"]
is_qdii, is_stock = fund.get("is_qdii"), fund.get("is_stock")
print(f"正在获取: {f_name}...")
if is_stock:
data = get_stock_data(f_code, f_name)
elif is_qdii:
# 此处省略 get_qdii_fund_data 定义，你可以直接贴回原代码中的该函数
data = None # 这里需调用你的 QDII 函数
else:
data = get_regular_fund_data(f_code, f_name)
all_report.append(format_fund_report(data, f_name, f_code, is_qdii, is_stock))
all_report.append("-" * 30)
final_report = "\n".join(all_report)
print(final_report)
send_wechat_notification(f"基金监控 {current_time}", final_report)
if __name__ == "__main__":
main()
0 条评论