import argparse
|
import csv
|
import os
|
from collections import Counter, defaultdict
|
from datetime import datetime, timezone, timedelta
|
from pathlib import Path
|
|
|
def read_csv(path):
|
with path.open("r", encoding="utf-8-sig", newline="") as handle:
|
return list(csv.DictReader(handle))
|
|
|
def top(counter, n=10):
|
return counter.most_common(n)
|
|
|
def safe_cell(value):
|
return (value or "").replace("\n", " ").strip()
|
|
|
def relative_link(target, base_dir):
|
return Path(os.path.relpath(target, base_dir)).as_posix()
|
|
|
COMPANY_STOPWORDS = {
|
"有色金属",
|
"能源金属",
|
"工业金属",
|
"基本金属",
|
"贵金属",
|
"小金属",
|
"金属新材料",
|
"中重稀土",
|
"矿产资源",
|
"自然资源",
|
"稀土资源",
|
"COMEX黄金",
|
"LME铜",
|
"LME铝",
|
"国家能源",
|
"中国新能源",
|
"基础化工",
|
"稀土冶炼",
|
"年全球稀土",
|
"全球稀土",
|
}
|
|
|
def main():
|
parser = argparse.ArgumentParser()
|
parser.add_argument("--project-root", default=".")
|
parser.add_argument("--outputs-dir", required=True)
|
parser.add_argument("--result-index", required=True)
|
parser.add_argument("--fact-files", nargs="+", required=True)
|
parser.add_argument("--company-files", nargs="+", required=True)
|
parser.add_argument("--snapshot", nargs="+", required=True)
|
args = parser.parse_args()
|
|
project_root = Path(args.project_root).resolve()
|
outputs_dir = project_root / args.outputs_dir
|
outputs_dir.mkdir(parents=True, exist_ok=True)
|
result_index = project_root / args.result_index
|
created_at = datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds")
|
|
facts = []
|
for file_name in args.fact_files:
|
path = project_root / file_name
|
if path.exists():
|
facts.extend(read_csv(path))
|
|
companies = []
|
for file_name in args.company_files:
|
path = project_root / file_name
|
if path.exists():
|
companies.extend(read_csv(path))
|
|
snapshot = []
|
for snapshot_file in args.snapshot:
|
path = project_root / snapshot_file
|
if path.exists():
|
snapshot.extend(read_csv(path))
|
|
metal_counter = Counter()
|
theme_counter = Counter()
|
metal_theme = defaultdict(Counter)
|
example_by_metal = defaultdict(list)
|
for row in facts:
|
metals = [tag for tag in row.get("metal_tags", "").split("|") if tag]
|
themes = [tag for tag in row.get("theme_tags", "").split("|") if tag]
|
for metal in metals:
|
metal_counter[metal] += 1
|
if len(example_by_metal[metal]) < 4:
|
example_by_metal[metal].append(row)
|
for theme in themes:
|
metal_theme[metal][theme] += 1
|
for theme in themes:
|
theme_counter[theme] += 1
|
|
company_counter = Counter()
|
company_examples = {}
|
for row in companies:
|
name = row.get("company_name", "")
|
if (
|
not name
|
or len(name) > 14
|
or name in COMPANY_STOPWORDS
|
or name.isdigit()
|
or "本报告" in name
|
or "所有材料" in name
|
):
|
continue
|
company_counter[name] += 1
|
company_examples.setdefault(name, row)
|
|
by_metal_snapshot = defaultdict(list)
|
for row in snapshot:
|
by_metal_snapshot[row.get("metal", "")].append(row)
|
|
industry_lines = [
|
"# 有色行业视图草稿",
|
"",
|
"状态：DRAFT_FOR_REVIEW",
|
f"生成时间：{created_at}",
|
"",
|
"## 证据边界",
|
"",
|
"本文档由 BATCH-001 与 BATCH-003 的规则抽取事实句、BATCH-003 转换/抽取 manifest、外部价格库存快照共同生成。当前尚未完成关键事实段落、表格定位和人工复核，因此只作为核心文档草稿，不输出正式行业结论或交易建议。",
|
"",
|
"## 材料覆盖",
|
"",
|
f"- 规则抽取事实句：{len(facts)} 条",
|
f"- 公司提及草表：{len(companies)} 条",
|
"- BATCH-003 转换状态：329 个补登记 raw 中 309 个已转换、20 个进入转换缺口。",
|
"",
|
"## 金属主题分布",
|
"",
|
"| 金属 | 事实句数量 | 主要主题 |",
|
"|---|---:|---|",
|
]
|
for metal, count in top(metal_counter, 12):
|
themes = ", ".join([f"{name}:{cnt}" for name, cnt in top(metal_theme[metal], 5)])
|
industry_lines.append(f"| {metal} | {count} | {themes} |")
|
industry_lines.extend(["", "## 初步行业读法", ""])
|
for metal, _ in top(metal_counter, 8):
|
themes = [name for name, _ in top(metal_theme[metal], 4)]
|
industry_lines.append(f"### {metal}")
|
industry_lines.append("")
|
industry_lines.append(f"- 抽取材料中，{metal}相关句子主要集中在：{', '.join(themes) if themes else '待复核'}。")
|
if example_by_metal[metal]:
|
industry_lines.append("- 待复核样例：")
|
for row in example_by_metal[metal][:2]:
|
industry_lines.append(f" - {safe_cell(row.get('evidence_text'))}（{row.get('doc_id')}，{row.get('source_location')}）")
|
industry_lines.append("")
|
industry_lines.extend(
|
[
|
"## 下一步复核点",
|
"",
|
"1. 对价格、库存、供给、项目投产和政策相关事实补段落或表格定位。",
|
"2. 将低置信度公司提及草表收敛为公司证据卡，补业务占比、资源/产能、成本位置和触发/失效条件。",
|
"3. 外部价格库存快照需要按来源继续补 SHFE、COMEX、SMM 等可复核数值。",
|
]
|
)
|
|
market_lines = [
|
"# 有色市场视图草稿",
|
"",
|
"状态：DRAFT_FOR_REVIEW",
|
f"生成时间：{created_at}",
|
"",
|
"## 证据边界",
|
"",
|
"本文档仅汇总公开外部快照和研报抽取中的市场相关句子。涉及实时行情、新闻、市场表现或外部信息的内容按 supplement/evidence 分账，不作为正式交易判断。",
|
"",
|
"## 外部价格库存快照",
|
"",
|
"| 金属 | 指标 | 数值 | 单位 | 市场/地区 | 来源 |",
|
"|---|---|---:|---|---|---|",
|
]
|
for row in snapshot:
|
market_lines.append(
|
f"| {row.get('metal')} | {row.get('indicator')} | {row.get('value')} | {row.get('unit')} | {row.get('market_or_region')} | {row.get('source_name')} |"
|
)
|
market_lines.extend(["", "## 市场变量分布", "", "| 主题 | 事实句数量 |", "|---|---:|"])
|
for theme, count in top(theme_counter, 12):
|
market_lines.append(f"| {theme} | {count} |")
|
market_lines.extend(
|
[
|
"",
|
"## 观察框架",
|
"",
|
"- 工业金属：优先跟踪价格、库存、冶炼/矿端供应、下游开工和宏观需求预期之间是否一致。",
|
"- 贵金属：区分金银价格的宏观利率/美元逻辑与材料端白银需求逻辑。",
|
"- 能源金属：锂、镍、钴需要把价格方向与库存、排产、产能释放和政策扰动分开验证。",
|
"- 稀土/小金属：重点核对供给管制、出口、库存和高端制造需求，不用单一价格点推导结论。",
|
"",
|
"## 缺口",
|
"",
|
"1. SHFE 库存与国内现货价格仍需补可复核数值入口。",
|
"2. 锂和稀土的部分价格为补充口径，仍需 SMM 或交易所/行业协会直接数值复核。",
|
"3. 市场异常表现、新闻事件和政策事件需要后续按 darkline 分流形成独立证据链。",
|
]
|
)
|
|
company_lines = [
|
"# 有色公司视图草稿",
|
"",
|
"状态：DRAFT_FOR_REVIEW",
|
f"生成时间：{created_at}",
|
"",
|
"## 证据边界",
|
"",
|
"公司视图当前来自规则公司名扫描，置信度较低，只用于确定后续证据卡优先级。不得据此输出交易指令、收益承诺或无证据强结论。",
|
"",
|
"## 公司提及频次",
|
"",
|
"| 公司名片段 | 提及次数 | 样例定位 | 样例上下文 |",
|
"|---|---:|---|---|",
|
]
|
for name, count in top(company_counter, 40):
|
row = company_examples[name]
|
company_lines.append(
|
f"| {name} | {count} | {row.get('doc_id')} / {row.get('source_location')} | {safe_cell(row.get('context_snippet'))[:120]} |"
|
)
|
company_lines.extend(
|
[
|
"",
|
"## 后续公司证据卡字段",
|
"",
|
"1. 公司简称、证券代码、主营金属、核心资产或业务环节。",
|
"2. 资源量、产能、成本位置、扩产/投产节奏、价格敏感性。",
|
"3. 观察条件、触发条件、失效条件和主要风险。",
|
"4. 原文页码/段落/表格定位、外部公告或数据来源、review_status。",
|
]
|
)
|
|
industry_path = outputs_dir / "nonferrous_industry_view_draft.md"
|
market_path = outputs_dir / "nonferrous_market_view_draft.md"
|
company_path = outputs_dir / "nonferrous_company_view_draft.md"
|
industry_path.write_text("\n".join(industry_lines) + "\n", encoding="utf-8")
|
market_path.write_text("\n".join(market_lines) + "\n", encoding="utf-8")
|
company_path.write_text("\n".join(company_lines) + "\n", encoding="utf-8")
|
|
result_lines = [
|
"# ANA-YS-INDUSTRY-001 结果入口",
|
"",
|
"状态：DRAFT_FOR_REVIEW",
|
f"更新时间：{created_at}",
|
"",
|
"## 核心文档",
|
"",
|
f"1. [有色行业视图草稿]({relative_link(industry_path, result_index.parent)})",
|
f"2. [有色市场视图草稿]({relative_link(market_path, result_index.parent)})",
|
f"3. [有色公司视图草稿]({relative_link(company_path, result_index.parent)})",
|
"",
|
"## 关键过程产物",
|
"",
|
"- `ana-data/cases/有色案例/ANA-YS-INDUSTRY-001/outputs/batch003_conversion_summary.md`",
|
"- `ana-data/cases/有色案例/ANA-YS-INDUSTRY-001/outputs/batch003_extract001_summary.md`",
|
"- `ana-data/cases/有色案例/ANA-YS-INDUSTRY-001/outputs/key_fact_review_batch003_pass001_summary.md`",
|
"- `ana-data/cases/有色案例/ANA-YS-INDUSTRY-001/outputs/company_evidence_card_batch003_pass001_summary.md`",
|
"- `ana-data/cases/有色案例/ANA-YS-INDUSTRY-001/outputs/price_inventory_supplement_pass002_summary.md`",
|
"- `ana-data/cases/有色案例/supplement/nonferrous_price_inventory_snapshot_20260625.csv`",
|
"- `ana-data/cases/有色案例/supplement/nonferrous_price_inventory_supplement_pass002_20260625.csv`",
|
"- `ana-data/cases/有色案例/manifest/conversion_gap_batch003.csv`",
|
"",
|
"## 限制",
|
"",
|
"BATCH-001 与 BATCH-003 关键事实尚未完成段落/表格定位和人工复核。对外正式结论、公司投资读法或交付前，必须提交执行审核或输出审核。",
|
]
|
result_index.parent.mkdir(parents=True, exist_ok=True)
|
result_index.write_text("\n".join(result_lines) + "\n", encoding="utf-8")
|
|
print(f"industry={industry_path.relative_to(project_root).as_posix()}")
|
print(f"market={market_path.relative_to(project_root).as_posix()}")
|
print(f"company={company_path.relative_to(project_root).as_posix()}")
|
print(f"result_index={result_index.relative_to(project_root).as_posix()}")
|
|
|
if __name__ == "__main__":
|
main()
|
