200 lines
7.4 KiB
Python
200 lines
7.4 KiB
Python
"""业绩曲线数据解析与图表生成。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import hashlib
|
|
from datetime import date, datetime
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def parse_performance_file(path: str | Path) -> dict[str, Any]:
|
|
source = Path(path)
|
|
if source.suffix.lower() == ".csv":
|
|
with source.open("r", encoding="utf-8-sig", newline="") as stream:
|
|
rows = list(csv.DictReader(stream))
|
|
elif source.suffix.lower() in {".xlsx", ".xlsm"}:
|
|
from openpyxl import load_workbook # type: ignore[import-untyped]
|
|
|
|
# read_only 模式的工作簿会持有文件句柄，必须显式关闭；否则上传后的
|
|
# 业绩文件一直被后端进程占用，运维侧无法移动或删除该文件。
|
|
workbook = load_workbook(source, read_only=True, data_only=True)
|
|
try:
|
|
values = list(workbook.active.values)
|
|
finally:
|
|
workbook.close()
|
|
if not values:
|
|
return {"columns": [], "rows": [], "row_count": 0}
|
|
headers = [str(value or "").strip() for value in values[0]]
|
|
rows = [
|
|
{headers[index]: value for index, value in enumerate(row) if index < len(headers)}
|
|
for row in values[1:]
|
|
]
|
|
else:
|
|
raise ValueError("业绩数据只支持 CSV、XLSX 或 XLSM")
|
|
normalized = [_normalize_row(row) for row in rows if any(row.values())]
|
|
columns = list(normalized[0]) if normalized else []
|
|
result = {"columns": columns, "rows": normalized, "row_count": len(normalized)}
|
|
_validate_performance_data(result)
|
|
return result
|
|
|
|
|
|
def create_performance_chart(
|
|
data: dict[str, Any], output_path: str | Path, *, title: str
|
|
) -> str:
|
|
import matplotlib
|
|
import matplotlib.dates as mdates
|
|
from matplotlib import font_manager
|
|
|
|
matplotlib.use("Agg")
|
|
import matplotlib.pyplot as plt
|
|
|
|
# 优先使用部署机上的中文字体，避免图表标题和图例缺字。
|
|
font_candidates = (
|
|
Path(r"C:\Windows\Fonts\msyh.ttc"),
|
|
Path(r"C:\Windows\Fonts\simhei.ttf"),
|
|
Path("/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc"),
|
|
Path("/usr/share/fonts/truetype/noto/NotoSansCJK-Regular.ttc"),
|
|
Path("/usr/share/fonts/truetype/wqy/wqy-zenhei.ttc"),
|
|
)
|
|
chinese_font = next((item for item in font_candidates if item.exists()), None)
|
|
if chinese_font is not None:
|
|
font_manager.fontManager.addfont(str(chinese_font))
|
|
font_name = font_manager.FontProperties(fname=str(chinese_font)).get_name()
|
|
matplotlib.rcParams["font.family"] = font_name
|
|
matplotlib.rcParams["axes.unicode_minus"] = False
|
|
|
|
rows = data.get("rows", [])
|
|
if not rows:
|
|
raise ValueError("业绩曲线没有可用数据")
|
|
date_key = _find_key(data, ("date", "日期", "净值日期", "统计日期"))
|
|
series = (
|
|
("product_return", "产品", "#1f4e79"),
|
|
("benchmark_return", "业绩比较基准", "#7f8c8d"),
|
|
("manager_representative_return", "基金经理代表产品", "#b5651d"),
|
|
)
|
|
fig, axis = plt.subplots(figsize=(10, 4.8), dpi=160)
|
|
plotted = 0
|
|
for key, label, color in series:
|
|
actual_key = _find_key(data, (key, _chinese_key(key)))
|
|
if actual_key is None:
|
|
continue
|
|
values = [_number(row.get(actual_key)) for row in rows]
|
|
dates = [_parse_date(row.get(date_key)) for row in rows]
|
|
pairs = [
|
|
(day, value)
|
|
for day, value in zip(dates, values, strict=True)
|
|
if day and value is not None
|
|
]
|
|
if not pairs:
|
|
continue
|
|
axis.plot(
|
|
[
|
|
float(mdates.date2num(item[0])) # type: ignore[no-untyped-call]
|
|
for item in pairs
|
|
],
|
|
[item[1] for item in pairs],
|
|
label=label, color=color, linewidth=1.8)
|
|
plotted += 1
|
|
if not plotted:
|
|
plt.close(fig)
|
|
raise ValueError("业绩曲线缺少可绘制序列")
|
|
axis.set_title(title)
|
|
axis.set_ylabel("累计收益 / 指数化净值")
|
|
axis.xaxis_date()
|
|
axis.grid(alpha=0.25)
|
|
axis.legend()
|
|
fig.autofmt_xdate()
|
|
output = Path(output_path)
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
fig.savefig(output, bbox_inches="tight")
|
|
plt.close(fig)
|
|
return str(output)
|
|
|
|
|
|
def file_hash(payload: bytes) -> str:
|
|
return hashlib.sha256(payload).hexdigest()
|
|
|
|
|
|
def _normalize_row(row: dict[Any, Any]) -> dict[str, Any]:
|
|
# openpyxl 把日期单元格读成 datetime，写进 MySQL JSON 列会直接抛
|
|
# "Object of type datetime is not JSON serializable"（表现为上传业绩文件 500）。
|
|
# 这里统一转成 JSON 安全类型：日期转 YYYY-MM-DD 字符串（_parse_date 可再次解析），
|
|
# Decimal 转 float，保证 CSV 和 XLSX 两种来源产出的结构一致。
|
|
return {
|
|
str(key).strip(): _json_safe(value)
|
|
for key, value in row.items()
|
|
if key is not None
|
|
}
|
|
|
|
|
|
def _json_safe(value: Any) -> Any:
|
|
if isinstance(value, datetime):
|
|
return value.date().strftime("%Y-%m-%d")
|
|
if isinstance(value, date):
|
|
return value.strftime("%Y-%m-%d")
|
|
if isinstance(value, Decimal):
|
|
return float(value)
|
|
return value
|
|
|
|
|
|
def _find_key(data: dict[str, Any], candidates: tuple[str, ...]) -> str | None:
|
|
columns = {str(column).strip() for column in data.get("columns", [])}
|
|
for candidate in candidates:
|
|
if candidate in columns:
|
|
return candidate
|
|
return None
|
|
|
|
|
|
def _validate_performance_data(data: dict[str, Any]) -> None:
|
|
if not data.get("rows"):
|
|
raise ValueError("业绩数据没有有效数据行")
|
|
date_key = _find_key(data, ("date", "日期", "净值日期", "统计日期"))
|
|
if date_key is None:
|
|
raise ValueError("业绩数据必须包含日期列")
|
|
parsed_dates = [_parse_date(row.get(date_key)) for row in data["rows"]]
|
|
if any(item is None for item in parsed_dates):
|
|
raise ValueError("业绩数据存在无法识别的日期")
|
|
valid_dates = [item for item in parsed_dates if item is not None]
|
|
if len(set(valid_dates)) != len(valid_dates):
|
|
raise ValueError("业绩数据存在重复日期")
|
|
if valid_dates != sorted(valid_dates):
|
|
raise ValueError("业绩数据必须按日期升序排列")
|
|
series_keys = (
|
|
("product_return", "产品收益率"),
|
|
("benchmark_return", "业绩比较基准收益率"),
|
|
("manager_representative_return", "基金经理代表产品收益率"),
|
|
)
|
|
if not any(_find_key(data, pair) for pair in series_keys):
|
|
raise ValueError("业绩数据至少需要一条可绘制曲线")
|
|
|
|
|
|
def _chinese_key(key: str) -> str:
|
|
return {
|
|
"product_return": "产品收益率",
|
|
"benchmark_return": "业绩比较基准收益率",
|
|
"manager_representative_return": "基金经理代表产品收益率",
|
|
}.get(key, key)
|
|
|
|
|
|
def _number(value: object) -> float | None:
|
|
try:
|
|
return float(str(value).replace("%", "").replace(",", "").strip())
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _parse_date(value: object) -> date | None:
|
|
if isinstance(value, datetime):
|
|
return value.date()
|
|
if isinstance(value, date):
|
|
return value
|
|
for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y%m%d"):
|
|
try:
|
|
return datetime.strptime(str(value), fmt).date()
|
|
except (TypeError, ValueError):
|
|
continue
|
|
return None
|