当前位置：首页 > news >正文

基金估值工具

news 2026/3/27 10:20:03

from flask import Flask, render_template, request, jsonify import requests import json import re import time from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry app = Flask(__name__) # 配置日志 import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 创建带重试机制的请求会话（增强请求头） def create_session(): """创建稳定的请求会话，包含完整的浏览器请求头""" session = requests.Session() retry = Retry( total=5, backoff_factor=2, status_forcelist=[500, 502, 503, 504], allowed_methods=["GET"] ) session.mount("http://", HTTPAdapter(max_retries=retry)) session.mount("https://", HTTPAdapter(max_retries=retry)) # 更完整的请求头，模拟真实浏览器 session.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Referer": "https://fund.eastmoney.com/", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Accept-Encoding": "gzip, deflate, br", "Connection": "keep-alive", "Upgrade-Insecure-Requests": "1", "Cache-Control": "max-age=0" }) return session class FundValuation: def __init__(self): self.session = create_session() def _extract_numeric_value(self, value): """提取数值，处理百分比和空值""" if not value or value == '': return '0' # 移除百分号并转换为数值 str_value = str(value).replace('%', '').strip() try: # 保留4位小数 return f"{float(str_value):.4f}" except: return '0' def _get_1234567_fund(self, fund_code): """1234567基金网数据源（优先）""" try: url = f"http://fundgz.1234567.com.cn/js/{fund_code}.js" response = self.session.get(url, timeout=10) response.raise_for_status() # 检查HTTP状态码 # 清理返回数据（处理可能的多余字符） raw_data = response.text.strip() # 提取JSON部分（兼容不同格式） json_match = re.search(r'jsonpgz\((.*)\);', raw_data) if not json_match: raise ValueError("未找到JSON数据") fund_info = json.loads(json_match.group(1)) # 基础数据 + 扩展字段（默认值） result = { "code": fund_info["fundcode"], "name": fund_info["name"], "current_value": fund_info["gsz"], # 实时估值 "current_rate": f"{fund_info['gszzl']}%", # 实时涨幅 "nav": fund_info.get("dwjz", "暂无数据"), # 单位净值 "accum_nav": fund_info.get("ljjz", "暂无数据"), # 累计净值 "day_rate": "暂无数据", # 日涨跌幅 "time": fund_info["gztime"], "source": "1234567基金网", "status": "success", # 新增收益率数据 "yield_1day": "暂无数据", "yield_1week": "暂无数据", "yield_1month": "暂无数据", "yield_3month": "暂无数据", "yield_6month": "暂无数据", "yield_1year": "暂无数据", "yield_2year": "暂无数据", "yield_3year": "暂无数据", "yield_since_inception": "暂无数据", # 成立以来收益率 "fund_type": "暂无数据", # 基金类型 "fund_company": "暂无数据" # 基金公司 } return result except Exception as e: logger.warning(f"1234567基金网接口失败: {str(e)}") return {"status": "failed", "error": str(e)} def _get_eastmoney_fund(self, fund_code): """东方财富网数据源（备用1）- 增强版，获取更多数据""" try: # 1. 获取基础估值数据 url = f"https://fund.eastmoney.com/pingzhongdata/{fund_code}.js" response = self.session.get(url, timeout=10) response.raise_for_status() response.encoding = "utf-8" raw_data = response.text.strip() # 提取基础数据 fund_data = {} # 匹配估值相关数据 gsz_match = re.search(r'gsz\s*:\s*"([^"]+)"', raw_data) gszzl_match = re.search(r'gszzl\s*:\s*"([^"]+)"', raw_data) gztime_match = re.search(r'gztime\s*:\s*"([^"]+)"', raw_data) name_match = re.search(r'fS_name\s*:\s*"([^"]+)"', raw_data) # 2. 获取详细数据（单位净值、收益率等） detail_url = f"https://fund.eastmoney.com/f10/F10DataApi.aspx?type=lsjz&code={fund_code}&page=1&per=1" detail_response = self.session.get(detail_url, timeout=10) detail_response.raise_for_status() detail_response.encoding = "utf-8" # 提取单位净值和累计净值 nav_match = re.search(r'<td class="tor bold">(.*?)</td>', detail_response.text) accum_nav_match = re.search(r'<td class="tor bold">(.*?)</td>.*?<td class="tor bold">(.*?)</td>', detail_response.text, re.DOTALL) # 3. 获取收益率数据 yield_url = f"https://fund.eastmoney.com/{fund_code}.html" yield_response = self.session.get(yield_url, timeout=10) yield_response.raise_for_status() yield_response.encoding = "utf-8" # 提取各类收益率 yield_patterns = { "yield_1day": r'(.*?).*?日涨跌幅', "yield_1week": r'(.*?).*?近1周', "yield_1month": r'(.*?).*?近1月', "yield_3month": r'(.*?).*?近3月', "yield_6month": r'(.*?).*?近6月', "yield_1year": r'(.*?).*?近1年', "yield_2year": r'(.*?).*?近2年', "yield_3year": r'(.*?).*?近3年', "yield_since_inception": r'(.*?).*?成立以来', "fund_type": r'<div class="fundInfoItem"><label>基金类型：</label>(.*?)</div>', "fund_company": r'<div class="fundInfoItem"><label>基金公司：</label>.*?<a.*?>(.*?)</a>.*?</div>' } yield_data = {} for key, pattern in yield_patterns.items(): match = re.search(pattern, yield_response.text, re.DOTALL) yield_data[key] = match.group(1).strip() if match else "暂无数据" # 构建完整结果 result = { "code": fund_code, "name": name_match.group(1).strip() if name_match else "未知名称", "current_value": gsz_match.group(1) if gsz_match else "暂无数据", "current_rate": f"{gszzl_match.group(1)}%" if gszzl_match else "0%", "nav": nav_match.group(1).strip() if nav_match else "暂无数据", "accum_nav": accum_nav_match.group(2).strip() if accum_nav_match else "暂无数据", "day_rate": yield_data["yield_1day"], "time": gztime_match.group(1) if gztime_match else "暂无数据", "source": "东方财富网", "status": "success", # 新增的收益率数据 "yield_1day": yield_data["yield_1day"], "yield_1week": yield_data["yield_1week"], "yield_1month": yield_data["yield_1month"], "yield_3month": yield_data["yield_3month"], "yield_6month": yield_data["yield_6month"], "yield_1year": yield_data["yield_1year"], "yield_2year": yield_data["yield_2year"], "yield_3year": yield_data["yield_3year"], "yield_since_inception": yield_data["yield_since_inception"], "fund_type": yield_data["fund_type"], "fund_company": yield_data["fund_company"] } return result except Exception as e: logger.warning(f"东方财富网接口失败: {str(e)}") return {"status": "failed", "error": str(e)} def _get_1234567fund_fund(self, fund_code): """天天基金网数据源（备用2）- 增强版""" try: url = f"https://fund.eastmoney.com/{fund_code}.html" response = self.session.get(url, timeout=10) response.raise_for_status() response.encoding = "utf-8" # 提取基础数据 name_match = re.search(r'<div class="fundDetail-tit">.*?>(.*?)</div>', response.text, re.DOTALL) value_match = re.search(r'(.*?)', response.text) rate_match = re.search(r'(.*?)', response.text) time_match = re.search(r'(.*?)', response.text) # 提取单位净值和累计净值 nav_match = re.search(r'<dt>单位净值\(.*?\)</dt>.*?<dd class="dataNums">(.*?)</dd>', response.text, re.DOTALL) accum_nav_match = re.search(r'<dt>累计净值</dt>.*?<dd class="dataNums">(.*?)</dd>', response.text, re.DOTALL) # 提取收益率数据 yield_patterns = { "yield_1day": r'(.*?).*?日涨跌幅', "yield_1week": r'(.*?).*?近1周', "yield_1month": r'(.*?).*?近1月', "yield_3month": r'(.*?).*?近3月', "yield_6month": r'(.*?).*?近6月', "yield_1year": r'(.*?).*?近1年', "yield_2year": r'(.*?).*?近2年', "yield_3year": r'(.*?).*?近3年', "yield_since_inception": r'(.*?).*?成立以来', "fund_type": r'<div class="fundInfoItem"><label>基金类型：</label>(.*?)</div>', "fund_company": r'<div class="fundInfoItem"><label>基金公司：</label>.*?<a.*?>(.*?)</a>.*?</div>' } yield_data = {} for key, pattern in yield_patterns.items(): match = re.search(pattern, response.text, re.DOTALL) yield_data[key] = match.group(1).strip() if match else "暂无数据" if not all([name_match, value_match, rate_match, time_match]): raise ValueError("部分基础数据提取失败") return { "code": fund_code, "name": name_match.group(1).strip(), "current_value": value_match.group(1).strip(), "current_rate": f"{rate_match.group(1).strip()}%", "nav": nav_match.group(1).strip() if nav_match else "暂无数据", "accum_nav": accum_nav_match.group(1).strip() if accum_nav_match else "暂无数据", "day_rate": yield_data["yield_1day"], "time": time_match.group(1).strip(), "source": "天天基金网", "status": "success", # 新增数据 "yield_1day": yield_data["yield_1day"], "yield_1week": yield_data["yield_1week"], "yield_1month": yield_data["yield_1month"], "yield_3month": yield_data["yield_3month"], "yield_6month": yield_data["yield_6month"], "yield_1year": yield_data["yield_1year"], "yield_2year": yield_data["yield_2year"], "yield_3year": yield_data["yield_3year"], "yield_since_inception": yield_data["yield_since_inception"], "fund_type": yield_data["fund_type"], "fund_company": yield_data["fund_company"] } except Exception as e: logger.warning(f"天天基金网接口失败: {str(e)}") return {"status": "failed", "error": str(e)} def get_fund_value(self, fund_code): """获取单只基金估值（三级备用数据源）""" # 校验基金代码格式 if not fund_code.isdigit() or len(fund_code) not in [5, 6]: return { "code": fund_code, "name": "无效代码", "current_value": "暂无数据", "current_rate": "暂无数据", "nav": "暂无数据", "accum_nav": "暂无数据", "day_rate": "暂无数据", "time": "暂无数据", "source": "无", "status": "failed", "error": "基金代码必须为5-6位数字", # 新增字段默认值 "yield_1day": "暂无数据", "yield_1week": "暂无数据", "yield_1month": "暂无数据", "yield_3month": "暂无数据", "yield_6month": "暂无数据", "yield_1year": "暂无数据", "yield_2year": "暂无数据", "yield_3year": "暂无数据", "yield_since_inception": "暂无数据", "fund_type": "暂无数据", "fund_company": "暂无数据" } # 第1优先级：1234567基金网 result = self._get_1234567_fund(fund_code) if result["status"] == "success": return result # 第2优先级：东方财富网 result = self._get_eastmoney_fund(fund_code) if result["status"] == "success": return result # 第3优先级：天天基金网（直接爬取页面） result = self._get_1234567fund_fund(fund_code) if result["status"] == "success": return result # 所有数据源都失败 return { "code": fund_code, "name": "未知基金", "current_value": "暂无数据", "current_rate": "暂无数据", "nav": "暂无数据", "accum_nav": "暂无数据", "day_rate": "暂无数据", "time": "暂无数据", "source": "无", "status": "failed", "error": "所有数据源均查询失败（可能被反爬/非交易时间）", # 新增字段默认值 "yield_1day": "暂无数据", "yield_1week": "暂无数据", "yield_1month": "暂无数据", "yield_3month": "暂无数据", "yield_6month": "暂无数据", "yield_1year": "暂无数据", "yield_2year": "暂无数据", "yield_3year": "暂无数据", "yield_since_inception": "暂无数据", "fund_type": "暂无数据", "fund_company": "暂无数据" } def batch_get_fund_values(self, fund_codes): """批量获取多只基金估值""" results = [] for code in fund_codes: code = code.strip() if code: # 跳过空值 result = self.get_fund_value(code) results.append(result) time.sleep(0.5) # 缩短间隔，同时降低反爬风险 return results # 初始化基金估值工具 fund_tool = FundValuation() # 路由：首页 @app.route("/", methods=["GET", "POST"]) def index(): if request.method == "POST": # 获取用户输入的基金代码 fund_input = request.form.get("fund_codes", "") # 分割多个基金代码（支持逗号、空格、换行、制表符） fund_codes = re.split(r'[,，\s\n\t]+', fund_input) fund_codes = [code for code in fund_codes if code.strip()] # 批量查询基金估值 fund_results = fund_tool.batch_get_fund_values(fund_codes) return render_template("index.html", results=fund_results, input_codes=fund_input) # GET请求：展示空页面 return render_template("index.html", results=[], input_codes="") # 路由：API接口（供前端异步调用） @app.route("/api/fund", methods=["POST"]) def api_fund(): data = request.get_json() fund_codes = data.get("fund_codes", []) if not isinstance(fund_codes, list): return jsonify({"status": "error", "message": "参数格式错误"}), 400 # 过滤空值 fund_codes = [code.strip() for code in fund_codes if code.strip()] results = fund_tool.batch_get_fund_values(fund_codes) return jsonify({"status": "success", "data": results}) # 新增：单个基金查询API @app.route("/api/fund/<fund_code>", methods=["GET"]) def api_single_fund(fund_code): """查询单只基金的详细信息""" result = fund_tool.get_fund_value(fund_code) if result["status"] == "success": return jsonify({"status": "success", "data": result}) else: return jsonify({"status": "error", "message": result["error"]}), 400 if __name__ == "__main__": # 关闭debug模式降低反爬风险，生产环境建议用False app.run(host="0.0.0.0", port=5000, debug=False)

查看全文

http://www.jsqmd.com/news/328103/