基金数据实战：从净值到持仓的全链路量化（第 8 篇）：基金经理画像：任职、业绩与在管规模

基金数据实战：从净值到持仓的全链路量化（第 8 篇）：基金经理画像：任职、业绩与在管规模

一、前言

前几篇我们把基金「产品」这一层的数据打通了——净值曲线、持仓明细，相当于把一只基金「长什么样」量了出来。但基金背后还有一个更关键的变量：基金经理。同一家公司、同一类策略，换一个经理，净值轨迹可能天差地别。

本篇把视角从「产品」切换到「人」，只用 jj/jjgk（基金概况，单只基金）这一个接口。注意：jjgk 是单只基金粒度，接口本身不提供任何「经理在管规模」「历任基金数」「费率性价比」等汇总指标。这些全部要靠我们用 Python 把多只基金的概况遍历后自研聚合出来。

衔接说明：本篇是「模块三·基金画像」的第一篇，从单只基金概况起步，自研经理维度聚合；下一篇（第 9 篇）将把本篇的 glr(管理人) 聚合推向「公司级」，完成「单只基金→经理→公司」的三级穿透。

二、接口字段解读

2.1 基金概况 jj/jjgk

https://api.biyingapi.com/jj/jjgk/{基金代码(如000001)}/{您的licence}

说明：jjgk 是单只基金接口，文档暴露字段均为「该基金本身」的概况，不含有任何经理汇总或公司业绩字段。本文所有「经理/公司」指标均为自研聚合，严格基于文档暴露字段，绝不假设清单外字段（如 tenure、return 等）。其中 zcgm/fegm/ljfh 等文档标注为 string，存库前需 float() 转换并容错；tzmb 等文本描述字段直接存储即可。

三、自研衍生指标 / 业务逻辑

接口只给裸字段，以下全部用 Python 手写：

四、数据表设计

CREATE TABLE IF NOT EXISTS fund_profile (
id INTEGER PRIMARY KEY AUTOINCREMENT,
dm TEXT, qc TEXT, jc TEXT, lx TEXT, pub TEXT, gm TEXT,
zcgm REAL, fegm REAL,
glr TEXT, tgr TEXT, jlr TEXT,
ljfh REAL,
glfv REAL, tgfl REAL, xsfl REAL,
rgfl REAL, sgfl REAL, shfl REAL,
bjjz TEXT, gzbd TEXT,
tzmb TEXT, tzln TEXT, tzfw TEXT, tzcl TEXT, fhzc TEXT, fxsytz TEXT,
collected_at TEXT,
UNIQUE(dm)
);

CREATE TABLE IF NOT EXISTS fund_manager_agg (
id INTEGER PRIMARY KEY AUTOINCREMENT,
jlr TEXT, — 基金经理人（分组键）
fund_count INTEGER, — 自研：历任基金数
total_zcgm REAL, — 自研：在管总规模
total_ljfh REAL, — 自研：累计分红合计
avg_fee REAL, — 自研：平均综合费率
unit_div REAL, — 自研：单位规模累计分红
collected_at TEXT,
UNIQUE(jlr)
);

CREATE TABLE IF NOT EXISTS fund_company_agg (
id INTEGER PRIMARY KEY AUTOINCREMENT,
glr TEXT, — 基金管理人（公司分组键）
fund_count INTEGER, — 自研：旗下基金数
total_zcgm REAL, — 自研：公司合计在管规模
collected_at TEXT,
UNIQUE(glr)
);

五、完整可运行代码

import requests
import logging
import time
import pandas as pd
import sqlite3

# ========== 全局配置 ==========
LICENCE = "你的licence"
DB_PATH = "quant.db"
LOG_FILE = "quant_collect.log"
API_BASE = "https://api.biyingapi.com"

# ———- 日志初始化 ———-
logging.basicConfig(
filename=LOG_FILE,
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
filemode="a"
)
logger = logging.getLogger(__name__)

# ———- 交易所后缀透传（基金代码按官方原值透传） ———-
def with_exchange_suffix(code):
# 基金代码按官方原值透传，不做 A/B 后缀拼接，直接返回原值
return code

# ———- 带重试 HTTP 请求（复用） ———-
def biying_api_get_retry(url, timeout=15, max_retry=3):
for attempt in range(1, max_retry + 1):
try:
resp = requests.get(url, timeout=timeout)
if resp.status_code == 200:
return resp.json()
logger.warning(f"HTTP状态码异常：{resp.status_code}，第{attempt}次重试")
except Exception as e:
logger.warning(f"网络请求异常，第{attempt}次重试，错误信息：{str(e)}")
time.sleep(2)
logger.error("达到最大重试次数，接口请求失败")
return []

# ———- 数值安全转换（容错） ———-
def safe_float(value):
# zcgm/fegm/ljfh 等文档标注为 string，转换 float 时容错
if value is None:
return None
s = str(value).strip()
if s == "" or s.lower() in ("nan", "none"):
return None
try:
return float(s)
except (ValueError, TypeError):
return None

# ———- 单只基金概况 ———-
def fetch_profile(code):
code = with_exchange_suffix(code)
url = f"{API_BASE}/jj/jjgk/{code}/{LICENCE}"
data = biying_api_get_retry(url)
if isinstance(data, list) and data:
return data[0] if isinstance(data[0], dict) else {}
if isinstance(data, dict):
return data
return {}

# ———- 落库 fund_profile ———-
def save_profile(profile):
if not profile:
return
conn = sqlite3.connect(DB_PATH)
try:
dm = profile.get("dm")
if dm is not None:
conn.execute("DELETE FROM fund_profile WHERE dm = ?", (dm,))
row = {
"dm": profile.get("dm"),
"qc": profile.get("qc"),
"jc": profile.get("jc"),
"lx": profile.get("lx"),
"pub": profile.get("pub"),
"gm": profile.get("gm"),
"zcgm": safe_float(profile.get("zcgm")),
"fegm": safe_float(profile.get("fegm")),
"glr": profile.get("glr"),
"tgr": profile.get("tgr"),
"jlr": profile.get("jlr"),
"ljfh": safe_float(profile.get("ljfh")),
"glfv": safe_float(profile.get("glfv")),
"tgfl": safe_float(profile.get("tgfl")),
"xsfl": safe_float(profile.get("xsfl")),
"rgfl": safe_float(profile.get("rgfl")),
"sgfl": safe_float(profile.get("sgfl")),
"shfl": safe_float(profile.get("shfl")),
"bjjz": profile.get("bjjz"),
"gzbd": profile.get("gzbd"),
"tzmb": profile.get("tzmb"),
"tzln": profile.get("tzln"),
"tzfw": profile.get("tzfw"),
"tzcl": profile.get("tzcl"),
"fhzc": profile.get("fhzc"),
"fxsytz": profile.get("fxsytz"),
"collected_at": pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S"),
}
pd.DataFrame([row]).to_sql("fund_profile", conn, if_exists="append", index=False)
conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS uk_dm ON fund_profile(dm)")
conn.commit()
except Exception as e:
logger.error(f"落库基金概况失败：{str(e)}")
finally:
conn.close()

# ———- 自研：经理维度聚合 ———-
def aggregate_manager(profiles):
"""按 jlr(基金经理人) 分组，自研在管规模/历任只数/费率性价比/分红强度"""
rows = {}
for p in profiles:
jlr = p.get("jlr")
if jlr is None or str(jlr).strip() == "":
continue # jlr 为 None 时跳过聚合
zcgm = safe_float(p.get("zcgm"))
ljfh = safe_float(p.get("ljfh"))
glfv = safe_float(p.get("glfv"))
tgfl = safe_float(p.get("tgfl"))
xsfl = safe_float(p.get("xsfl"))
fees = [f for f in [glfv, tgfl, xsfl] if f is not None]
comp_fee = round(sum(fees), 4) if fees else None # 自研：综合费率
if jlr not in rows:
rows[jlr] = {"fund_count": 0, "total_zcgm": 0.0,
"total_ljfh": 0.0, "fee_sum": 0.0, "fee_n": 0}
r = rows[jlr]
r["fund_count"] += 1
if zcgm is not None:
r["total_zcgm"] += zcgm
if ljfh is not None:
r["total_ljfh"] += ljfh
if comp_fee is not None:
r["fee_sum"] += comp_fee
r["fee_n"] += 1
result = []
for jlr, r in rows.items():
avg_fee = (round(r["fee_sum"] / r["fee_n"], 4)
if r["fee_n"] > 0 else None) # 自研：平均综合费率
unit_div = (round(r["total_ljfh"] / r["total_zcgm"], 6)
if r["total_zcgm"] else None) # 自研：单位规模累计分红
result.append({
"jlr": jlr,
"fund_count": r["fund_count"], # 自研：历任基金数
"total_zcgm": round(r["total_zcgm"], 2), # 自研：在管总规模
"total_ljfh": round(r["total_ljfh"], 2), # 自研：累计分红合计
"avg_fee": avg_fee, # 自研：平均综合费率
"unit_div": unit_div, # 自研：单位规模累计分红
})
return result

# ———- 自研：公司维度预聚合 ———-
def aggregate_company(profiles):
"""按 glr(基金管理人) 分组，为第9篇铺垫"""
rows = {}
for p in profiles:
glr = p.get("glr")
if glr is None or str(glr).strip() == "":
continue
zcgm = safe_float(p.get("zcgm"))
if glr not in rows:
rows[glr] = {"fund_count": 0, "total_zcgm": 0.0}
r = rows[glr]
r["fund_count"] += 1
if zcgm is not None:
r["total_zcgm"] += zcgm
result = []
for glr, r in rows.items():
result.append({
"glr": glr,
"fund_count": r["fund_count"], # 自研：旗下基金数
"total_zcgm": round(r["total_zcgm"], 2), # 自研：公司合计在管规模
})
return result

# ———- 落库 fund_manager_agg（自研聚合） ———-
def save_manager_agg(agg):
if not agg:
return
conn = sqlite3.connect(DB_PATH)
try:
now = pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S")
df = pd.DataFrame(agg)
df["collected_at"] = now
keys = df["jlr"].tolist()
placeholders = ",".join("?" * len(keys))
conn.execute(
f"DELETE FROM fund_manager_agg WHERE jlr IN ({placeholders})", keys
)
df.to_sql("fund_manager_agg", conn, if_exists="append", index=False)
conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS uk_jlr ON fund_manager_agg(jlr)")
conn.commit()
except Exception as e:
logger.error(f"落库经理聚合失败：{str(e)}")
finally:
conn.close()

# ———- 落库 fund_company_agg（自研聚合） ———-
def save_company_agg(agg):
if not agg:
return
conn = sqlite3.connect(DB_PATH)
try:
now = pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S")
df = pd.DataFrame(agg)
df["collected_at"] = now
keys = df["glr"].tolist()
placeholders = ",".join("?" * len(keys))
conn.execute(
f"DELETE FROM fund_company_agg WHERE glr IN ({placeholders})", keys
)
df.to_sql("fund_company_agg", conn, if_exists="append", index=False)
conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS uk_glr ON fund_company_agg(glr)")
conn.commit()
except Exception as e:
logger.error(f"落库公司聚合失败：{str(e)}")
finally:
conn.close()

# ———- 演示入口（占位说明） ———-
def demo(fund_list=None):
# 以下仅为示例占位：用示例基金代码 000001 及若干占位代码，
# 演示「遍历多只基金 + 按 jlr/glr 聚合」，不假装真实跑出全市场数字
if fund_list is None:
fund_list = ["000001", "000002", "000003", "000004"]
profiles = []
for code in fund_list:
p = fetch_profile(code)
if p:
profiles.append(p)
save_profile(p)
mgr = aggregate_manager(profiles)
comp = aggregate_company(profiles)
print("\\n=== 经理维度聚合（自研） ===")
for m in mgr:
print(f"经理 {m['jlr']} | 在管 {m['fund_count']} 只 | "
f"总规模 {m['total_zcgm']} | 综合费率 {m['avg_fee']} | "
f"单位分红 {m['unit_div']}")
print("\\n=== 公司维度聚合（自研，为第9篇铺垫） ===")
for c in comp:
print(f"公司 {c['glr']} | 旗下 {c['fund_count']} 只 | 总规模 {c['total_zcgm']}")
save_manager_agg(mgr)
save_company_agg(comp)
print("\\n【仅为数据演示，不构成投资建议】已采集基金概况并完成经理/公司聚合")

if __name__ == "__main__":
demo()

六、业务关键点

七、拓展练习

八、下篇预告

下一篇《基金公司数据：公司维度聚合与对比》将把本篇的 glr(管理人) 聚合推向「公司级」——用 jjgk 的 glr 聚合全市场基金公司（旗下基金数、总在管规模、平均费率、平均分红），再用 js/gm/kfgm(总募集规模 zmjgm、最近总份额 zjzfe) 补充规模细节，并自研「公司规模集中度」「公司平均费率」「公司分红意愿」等指标，从「单只基金→经理→公司」三级穿透收口。

九、免责申明

免责申明：文中所有数据处理逻辑仅为编程演示，仅为数据演示，不构成投资建议。市场有风险，投资需谨慎。