496 lines
17 KiB
Python
496 lines
17 KiB
Python
"""财报深度解读 — 关键指标趋势、AI摘要、同行对比、异常预警。
|
||
|
||
功能：
|
||
1. 财报关键指标趋势
|
||
2. AI财报摘要
|
||
3. 同行对比
|
||
4. 财报异常预警
|
||
5. 财报发布日历
|
||
"""
|
||
import datetime as dt
|
||
import json
|
||
from typing import List, Dict, Any, Optional
|
||
from collections import defaultdict
|
||
import numpy as np
|
||
from sqlalchemy import select, and_, func, desc
|
||
|
||
from db import get_session
|
||
from models import FinancialReport, Security, StockMetric
|
||
import llm
|
||
|
||
|
||
def seed_sample_reports():
|
||
"""生成示例财报数据（用于演示）"""
|
||
with get_session() as s:
|
||
stocks = s.execute(
|
||
select(Security.code, Security.name).limit(50)
|
||
).all()
|
||
|
||
saved = 0
|
||
for code, name in stocks:
|
||
# 生成最近4个季度的财报
|
||
base_date = dt.date(2023, 12, 31)
|
||
|
||
for i in range(4):
|
||
report_date = base_date - dt.timedelta(days=i * 90)
|
||
publish_date = report_date + dt.timedelta(days=30)
|
||
|
||
# 检查是否已存在
|
||
exists = s.execute(
|
||
select(FinancialReport).where(
|
||
and_(
|
||
FinancialReport.code == code,
|
||
FinancialReport.report_date == report_date
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
|
||
if exists:
|
||
continue
|
||
|
||
# 模拟财务数据
|
||
base_revenue = np.random.uniform(10, 500)
|
||
growth = np.random.uniform(-20, 50)
|
||
|
||
report = FinancialReport(
|
||
code=code,
|
||
name=name,
|
||
report_date=report_date,
|
||
publish_date=publish_date,
|
||
report_type='Q' + str((report_date.month // 3) or 4),
|
||
revenue=round(base_revenue * (1 + i * 0.1), 2),
|
||
net_profit=round(base_revenue * np.random.uniform(0.05, 0.2), 2),
|
||
roe=round(np.random.uniform(5, 25), 2),
|
||
gross_margin=round(np.random.uniform(20, 60), 2),
|
||
revenue_growth=round(growth, 2),
|
||
profit_growth=round(growth + np.random.uniform(-10, 10), 2),
|
||
inventory=round(base_revenue * np.random.uniform(0.1, 0.3), 2),
|
||
receivable=round(base_revenue * np.random.uniform(0.15, 0.4), 2),
|
||
debt_ratio=round(np.random.uniform(30, 70), 2)
|
||
)
|
||
s.add(report)
|
||
saved += 1
|
||
|
||
s.commit()
|
||
|
||
return {"ok": True, "saved": saved}
|
||
|
||
|
||
def get_report_trend(code: str, periods: int = 8) -> Dict[str, Any]:
|
||
"""获取财报关键指标趋势
|
||
|
||
Args:
|
||
code: 股票代码
|
||
periods: 统计期数
|
||
|
||
Returns:
|
||
趋势数据
|
||
"""
|
||
with get_session() as s:
|
||
reports = s.execute(
|
||
select(FinancialReport)
|
||
.where(FinancialReport.code == code)
|
||
.order_by(desc(FinancialReport.report_date))
|
||
.limit(periods)
|
||
).scalars().all()
|
||
|
||
if not reports:
|
||
return {"ok": False, "msg": "暂无财报数据"}
|
||
|
||
# 反转顺序（从旧到新）
|
||
reports = list(reversed(reports))
|
||
|
||
# 提取趋势数据
|
||
dates = [r.report_date.isoformat() for r in reports]
|
||
|
||
trend_data = {
|
||
"revenue": [r.revenue for r in reports],
|
||
"net_profit": [r.net_profit for r in reports],
|
||
"roe": [r.roe for r in reports],
|
||
"gross_margin": [r.gross_margin for r in reports],
|
||
"revenue_growth": [r.revenue_growth for r in reports],
|
||
"profit_growth": [r.profit_growth for r in reports],
|
||
"debt_ratio": [r.debt_ratio for r in reports]
|
||
}
|
||
|
||
# 计算趋势（上升/下降/平稳）
|
||
def calc_trend(values):
|
||
if len(values) < 2:
|
||
return "平稳"
|
||
recent = np.mean(values[-2:])
|
||
previous = np.mean(values[:2]) if len(values) >= 4 else values[0]
|
||
change = (recent - previous) / previous if previous != 0 else 0
|
||
if change > 0.1:
|
||
return "上升"
|
||
elif change < -0.1:
|
||
return "下降"
|
||
else:
|
||
return "平稳"
|
||
|
||
trends = {
|
||
key: calc_trend(values)
|
||
for key, values in trend_data.items()
|
||
}
|
||
|
||
return {
|
||
"ok": True,
|
||
"code": code,
|
||
"name": reports[0].name,
|
||
"dates": dates,
|
||
"data": trend_data,
|
||
"trends": trends,
|
||
"latest": {
|
||
"revenue": reports[-1].revenue,
|
||
"net_profit": reports[-1].net_profit,
|
||
"roe": reports[-1].roe,
|
||
"gross_margin": reports[-1].gross_margin,
|
||
"revenue_growth": reports[-1].revenue_growth,
|
||
"profit_growth": reports[-1].profit_growth
|
||
}
|
||
}
|
||
|
||
|
||
def generate_ai_summary(code: str) -> Dict[str, Any]:
|
||
"""生成AI财报摘要
|
||
|
||
Args:
|
||
code: 股票代码
|
||
|
||
Returns:
|
||
AI摘要
|
||
"""
|
||
with get_session() as s:
|
||
# 获取最新财报
|
||
report = s.execute(
|
||
select(FinancialReport)
|
||
.where(FinancialReport.code == code)
|
||
.order_by(desc(FinancialReport.report_date))
|
||
.limit(1)
|
||
).scalar_one_or_none()
|
||
|
||
if not report:
|
||
return {"ok": False, "msg": "暂无财报数据"}
|
||
|
||
# 如果已有摘要，直接返回
|
||
if report.ai_summary:
|
||
return {
|
||
"ok": True,
|
||
"summary": report.ai_summary,
|
||
"report_date": report.report_date.isoformat()
|
||
}
|
||
|
||
# 构建提示词
|
||
prompt = f"""请用一句话总结以下财报数据（40字以内）：
|
||
|
||
公司：{report.name}（{report.code}）
|
||
报告期：{report.report_date}
|
||
营业收入：{report.revenue}亿元，同比增长{report.revenue_growth:+.1f}%
|
||
净利润：{report.net_profit}亿元，同比增长{report.profit_growth:+.1f}%
|
||
ROE：{report.roe}%
|
||
毛利率：{report.gross_margin}%
|
||
|
||
要求：
|
||
1. 一句话说明业绩是增长还是下降
|
||
2. 提及最亮眼或最担忧的指标
|
||
3. 给出简短评价（优秀/良好/一般/较差）
|
||
4. 不超过40字
|
||
|
||
示例：业绩稳步增长，ROE达20%创新高，盈利能力优秀。
|
||
"""
|
||
|
||
# 调用AI
|
||
if llm.enabled():
|
||
try:
|
||
summary = llm.ask(prompt, max_tokens=100)
|
||
# 保存摘要
|
||
report.ai_summary = summary
|
||
s.commit()
|
||
except Exception as e:
|
||
summary = f"营收{report.revenue}亿元({report.revenue_growth:+.1f}%)，净利润{report.net_profit}亿元({report.profit_growth:+.1f}%)"
|
||
else:
|
||
summary = f"营收{report.revenue}亿元({report.revenue_growth:+.1f}%)，净利润{report.net_profit}亿元({report.profit_growth:+.1f}%)"
|
||
|
||
return {
|
||
"ok": True,
|
||
"summary": summary,
|
||
"report_date": report.report_date.isoformat()
|
||
}
|
||
|
||
|
||
def compare_with_peers(code: str, sector: str = None) -> Dict[str, Any]:
|
||
"""同行对比
|
||
|
||
Args:
|
||
code: 股票代码
|
||
sector: 行业（可选）
|
||
|
||
Returns:
|
||
对比结果
|
||
"""
|
||
with get_session() as s:
|
||
# 获取目标股票最新财报
|
||
target = s.execute(
|
||
select(FinancialReport)
|
||
.where(FinancialReport.code == code)
|
||
.order_by(desc(FinancialReport.report_date))
|
||
.limit(1)
|
||
).scalar_one_or_none()
|
||
|
||
if not target:
|
||
return {"ok": False, "msg": "暂无财报数据"}
|
||
|
||
# 获取同行业股票（简化：随机选取）
|
||
peers = s.execute(
|
||
select(FinancialReport)
|
||
.where(
|
||
and_(
|
||
FinancialReport.code != code,
|
||
FinancialReport.report_date == target.report_date
|
||
)
|
||
)
|
||
.limit(20)
|
||
).scalars().all()
|
||
|
||
if not peers:
|
||
return {"ok": False, "msg": "暂无同行数据"}
|
||
|
||
# 计算行业均值
|
||
industry_avg = {
|
||
"roe": np.mean([p.roe for p in peers]),
|
||
"gross_margin": np.mean([p.gross_margin for p in peers]),
|
||
"revenue_growth": np.mean([p.revenue_growth for p in peers]),
|
||
"profit_growth": np.mean([p.profit_growth for p in peers]),
|
||
"debt_ratio": np.mean([p.debt_ratio for p in peers])
|
||
}
|
||
|
||
# 计算差异
|
||
comparison = {
|
||
"roe": {
|
||
"value": target.roe,
|
||
"industry_avg": round(industry_avg["roe"], 2),
|
||
"diff": round(target.roe - industry_avg["roe"], 2),
|
||
"better": target.roe > industry_avg["roe"]
|
||
},
|
||
"gross_margin": {
|
||
"value": target.gross_margin,
|
||
"industry_avg": round(industry_avg["gross_margin"], 2),
|
||
"diff": round(target.gross_margin - industry_avg["gross_margin"], 2),
|
||
"better": target.gross_margin > industry_avg["gross_margin"]
|
||
},
|
||
"revenue_growth": {
|
||
"value": target.revenue_growth,
|
||
"industry_avg": round(industry_avg["revenue_growth"], 2),
|
||
"diff": round(target.revenue_growth - industry_avg["revenue_growth"], 2),
|
||
"better": target.revenue_growth > industry_avg["revenue_growth"]
|
||
},
|
||
"profit_growth": {
|
||
"value": target.profit_growth,
|
||
"industry_avg": round(industry_avg["profit_growth"], 2),
|
||
"diff": round(target.profit_growth - industry_avg["profit_growth"], 2),
|
||
"better": target.profit_growth > industry_avg["profit_growth"]
|
||
},
|
||
"debt_ratio": {
|
||
"value": target.debt_ratio,
|
||
"industry_avg": round(industry_avg["debt_ratio"], 2),
|
||
"diff": round(target.debt_ratio - industry_avg["debt_ratio"], 2),
|
||
"better": target.debt_ratio < industry_avg["debt_ratio"] # 负债率越低越好
|
||
}
|
||
}
|
||
|
||
# 综合评分
|
||
better_count = sum(1 for v in comparison.values() if v["better"])
|
||
|
||
return {
|
||
"ok": True,
|
||
"code": code,
|
||
"name": target.name,
|
||
"report_date": target.report_date.isoformat(),
|
||
"comparison": comparison,
|
||
"better_count": better_count,
|
||
"total_metrics": len(comparison),
|
||
"conclusion": "优于行业" if better_count >= 3 else ("持平行业" if better_count == 2 else "弱于行业")
|
||
}
|
||
|
||
|
||
def detect_abnormalities(code: str) -> Dict[str, Any]:
|
||
"""财报异常预警
|
||
|
||
Args:
|
||
code: 股票代码
|
||
|
||
Returns:
|
||
异常预警
|
||
"""
|
||
with get_session() as s:
|
||
# 获取最近2期财报
|
||
reports = s.execute(
|
||
select(FinancialReport)
|
||
.where(FinancialReport.code == code)
|
||
.order_by(desc(FinancialReport.report_date))
|
||
.limit(2)
|
||
).scalars().all()
|
||
|
||
if len(reports) < 2:
|
||
return {"ok": False, "msg": "数据不足"}
|
||
|
||
current, previous = reports[0], reports[1]
|
||
|
||
warnings = []
|
||
|
||
# 1. 存货激增
|
||
if current.inventory > 0 and previous.inventory > 0:
|
||
inventory_growth = (current.inventory / previous.inventory - 1) * 100
|
||
if inventory_growth > 50:
|
||
warnings.append({
|
||
"type": "存货激增",
|
||
"severity": "high",
|
||
"description": f"存货增长{inventory_growth:.1f}%，可能存在滞销风险",
|
||
"current": current.inventory,
|
||
"previous": previous.inventory
|
||
})
|
||
|
||
# 2. 应收账款占比过高
|
||
receivable_ratio = current.receivable / current.revenue * 100 if current.revenue > 0 else 0
|
||
if receivable_ratio > 50:
|
||
warnings.append({
|
||
"type": "应收账款占比过高",
|
||
"severity": "medium",
|
||
"description": f"应收账款占营收{receivable_ratio:.1f}%，回款压力较大",
|
||
"ratio": round(receivable_ratio, 2)
|
||
})
|
||
|
||
# 3. 毛利率大幅下降
|
||
if current.gross_margin > 0 and previous.gross_margin > 0:
|
||
margin_change = current.gross_margin - previous.gross_margin
|
||
if margin_change < -5:
|
||
warnings.append({
|
||
"type": "毛利率大幅下降",
|
||
"severity": "high",
|
||
"description": f"毛利率下降{abs(margin_change):.1f}个百分点，盈利能力恶化",
|
||
"current": current.gross_margin,
|
||
"previous": previous.gross_margin
|
||
})
|
||
|
||
# 4. 资产负债率过高
|
||
if current.debt_ratio > 70:
|
||
warnings.append({
|
||
"type": "资产负债率过高",
|
||
"severity": "medium",
|
||
"description": f"资产负债率{current.debt_ratio}%，财务风险较高",
|
||
"value": current.debt_ratio
|
||
})
|
||
|
||
# 5. 增收不增利
|
||
if current.revenue_growth > 10 and current.profit_growth < 0:
|
||
warnings.append({
|
||
"type": "增收不增利",
|
||
"severity": "high",
|
||
"description": f"营收增长{current.revenue_growth:.1f}%，但净利润下降{abs(current.profit_growth):.1f}%",
|
||
"revenue_growth": current.revenue_growth,
|
||
"profit_growth": current.profit_growth
|
||
})
|
||
|
||
return {
|
||
"ok": True,
|
||
"code": code,
|
||
"name": current.name,
|
||
"report_date": current.report_date.isoformat(),
|
||
"warnings": warnings,
|
||
"risk_level": "高" if any(w["severity"] == "high" for w in warnings) else ("中" if warnings else "低")
|
||
}
|
||
|
||
|
||
def get_report_calendar(days: int = 30) -> Dict[str, Any]:
|
||
"""财报发布日历
|
||
|
||
Args:
|
||
days: 未来N天
|
||
|
||
Returns:
|
||
日历数据
|
||
"""
|
||
today = dt.date.today()
|
||
end_date = today + dt.timedelta(days=days)
|
||
|
||
with get_session() as s:
|
||
reports = s.execute(
|
||
select(FinancialReport)
|
||
.where(
|
||
and_(
|
||
FinancialReport.publish_date >= today,
|
||
FinancialReport.publish_date <= end_date
|
||
)
|
||
)
|
||
.order_by(FinancialReport.publish_date)
|
||
).scalars().all()
|
||
|
||
if not reports:
|
||
return {"ok": False, "msg": "暂无即将发布的财报"}
|
||
|
||
# 按日期分组
|
||
calendar = defaultdict(list)
|
||
for r in reports:
|
||
calendar[r.publish_date.isoformat()].append({
|
||
"code": r.code,
|
||
"name": r.name,
|
||
"report_date": r.report_date.isoformat(),
|
||
"report_type": r.report_type
|
||
})
|
||
|
||
return {
|
||
"ok": True,
|
||
"days": days,
|
||
"total": len(reports),
|
||
"calendar": dict(calendar)
|
||
}
|
||
|
||
|
||
def get_top_reports(metric: str = "roe", limit: int = 20) -> Dict[str, Any]:
|
||
"""获取财报排行榜
|
||
|
||
Args:
|
||
metric: 排序指标（roe/gross_margin/revenue_growth）
|
||
limit: 返回数量
|
||
|
||
Returns:
|
||
排行榜
|
||
"""
|
||
with get_session() as s:
|
||
# 获取最新一期的所有财报
|
||
latest_date = s.execute(
|
||
select(func.max(FinancialReport.report_date))
|
||
).scalar()
|
||
|
||
if not latest_date:
|
||
return {"ok": False, "msg": "暂无数据"}
|
||
|
||
# 根据指标排序
|
||
order_field = getattr(FinancialReport, metric)
|
||
|
||
reports = s.execute(
|
||
select(FinancialReport)
|
||
.where(FinancialReport.report_date == latest_date)
|
||
.order_by(desc(order_field))
|
||
.limit(limit)
|
||
).scalars().all()
|
||
|
||
results = []
|
||
for r in reports:
|
||
results.append({
|
||
"code": r.code,
|
||
"name": r.name,
|
||
"roe": r.roe,
|
||
"gross_margin": r.gross_margin,
|
||
"revenue_growth": r.revenue_growth,
|
||
"profit_growth": r.profit_growth,
|
||
metric: getattr(r, metric)
|
||
})
|
||
|
||
return {
|
||
"ok": True,
|
||
"metric": metric,
|
||
"report_date": latest_date.isoformat(),
|
||
"rankings": results
|
||
}
|
||
|