#!/usr/bin/env python3 """Fetch point-in-time A-share market structure from the iFinD HTTP API. Authentication is read from IFIND_REFRESH_TOKEN. Tokens are never written to disk. Raw API responses are stored as gzip-compressed JSON for auditability. """ from __future__ import annotations import argparse import csv import gzip import json import os import time import urllib.error import urllib.request from collections import defaultdict from dataclasses import dataclass from pathlib import Path from typing import Any, Iterable API_BASE = "https://quantapi.51ifind.com/api/v1" POOL_REPORT = "p03425" POOL_FIELDS = "p03291_f001,p03291_f002,p03291_f003,p03291_f004" @dataclass(frozen=True) class SnapshotDate: year: int market_date: str report_date: str SNAPSHOTS = ( SnapshotDate(1995, "1995-12-29", "1995-12-31"), SnapshotDate(2000, "2000-12-29", "2000-12-31"), SnapshotDate(2005, "2005-12-30", "2005-12-31"), SnapshotDate(2010, "2010-12-31", "2010-12-31"), SnapshotDate(2015, "2015-12-31", "2015-12-31"), SnapshotDate(2020, "2020-12-31", "2020-12-31"), SnapshotDate(2025, "2025-12-31", "2025-12-31"), ) METRICS = ( "ths_market_value_stock", "ths_current_mv_stock", "ths_free_float_mv_stock", "ths_the_csrc_industry_stock", "ths_the_csrc_industry_code_stock", "ths_the_new_csrc_industry_stock", "ths_the_new_csrc_industry_code_stock", "ths_china_securities_industry_stock", "ths_china_securities_indus_code_stock", "ths_np_atoopc_stock", "ths_operating_total_revenue_stock", "ths_total_equity_atoopc_stock", "ths_roe_stock", "ths_pe_ttm_stock", "ths_pb_mrq_stock", "ths_turnover_ratio_y_stock", "ths_fund_held_ratio_stock", ) FIELD_LABELS = { "ths_market_value_stock": "total_market_cap", "ths_current_mv_stock": "circulating_market_cap", "ths_free_float_mv_stock": "free_float_market_cap", "ths_the_csrc_industry_stock": "csrc_industry", "ths_the_csrc_industry_code_stock": "csrc_industry_code", "ths_the_new_csrc_industry_stock": "new_csrc_industry", "ths_the_new_csrc_industry_code_stock": "new_csrc_industry_code", "ths_china_securities_industry_stock": "csi_industry_2021", "ths_china_securities_indus_code_stock": "csi_industry_2021_code", "ths_np_atoopc_stock": "net_profit_parent", "ths_operating_total_revenue_stock": "operating_revenue", "ths_total_equity_atoopc_stock": "equity_parent", "ths_roe_stock": "roe", "ths_pe_ttm_stock": "pe_ttm", "ths_pb_mrq_stock": "pb_mrq", "ths_turnover_ratio_y_stock": "annual_turnover", "ths_fund_held_ratio_stock": "fund_held_ratio", } STABLE_INDUSTRIES = ( "能源", "原材料", "工业", "可选消费", "主要消费", "医药卫生", "金融地产", "信息技术", "通信服务", "公用事业", ) def post_json(path: str, payload: dict[str, Any] | None, headers: dict[str, str], retries: int = 4) -> dict[str, Any]: data = None if payload is None else json.dumps(payload, ensure_ascii=False).encode("utf-8") request = urllib.request.Request( f"{API_BASE}/{path}", data=data, headers={"Content-Type": "application/json", **headers}, method="POST" ) for attempt in range(retries): try: with urllib.request.urlopen(request, timeout=90) as response: result = json.loads(response.read().decode("utf-8")) if result.get("errorcode") not in (0, "0", None): raise RuntimeError(f"iFinD {path} failed: {result.get('errorcode')} {result.get('errmsg')}") return result except (urllib.error.URLError, TimeoutError, RuntimeError) as exc: if attempt == retries - 1: raise time.sleep(2**attempt) raise AssertionError("unreachable") def exchange_access_token(refresh_token: str) -> str: result = post_json("get_access_token", None, {"refresh_token": refresh_token}) candidates = [ result.get("access_token"), (result.get("data") or {}).get("access_token") if isinstance(result.get("data"), dict) else None, (result.get("tables") or {}).get("access_token") if isinstance(result.get("tables"), dict) else None, ] token = next((item for item in candidates if isinstance(item, str) and item), None) if not token: raise RuntimeError("iFinD token exchange succeeded but no access_token was returned") return token def write_raw(path: Path, value: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with gzip.open(path, "wt", encoding="utf-8") as handle: json.dump(value, handle, ensure_ascii=False, separators=(",", ":")) def first_value(value: Any) -> Any: if isinstance(value, list): return value[0] if value else None return value def chunks(items: list[str], size: int) -> Iterable[list[str]]: for offset in range(0, len(items), size): yield items[offset : offset + size] def fetch_stock_pool(access_token: str, snapshot: SnapshotDate, raw_dir: Path, refresh: bool) -> list[dict[str, str]]: raw_path = raw_dir / f"stock_pool_{snapshot.year}_{snapshot.market_date}.json.gz" if raw_path.exists() and not refresh: with gzip.open(raw_path, "rt", encoding="utf-8") as handle: response = json.load(handle) else: response = post_json( "data_pool", { "reportname": POOL_REPORT, "functionpara": { "date": snapshot.market_date.replace("-", ""), "blockname": "001005010", "iv_type": "allcontract", }, "outputpara": POOL_FIELDS, }, {"access_token": access_token}, ) write_raw(raw_path, response) table = response["tables"][0]["table"] codes = table["p03291_f002"] names = table["p03291_f003"] return [{"code": str(code), "name": str(name)} for code, name in zip(codes, names)] def metric_payload(codes: list[str], snapshot: SnapshotDate) -> dict[str, Any]: market_date = snapshot.market_date report_date = snapshot.report_date return { "codes": ",".join(codes), "indipara": [ {"indicator": "ths_market_value_stock", "indiparams": [market_date]}, {"indicator": "ths_current_mv_stock", "indiparams": [market_date]}, {"indicator": "ths_free_float_mv_stock", "indiparams": [market_date]}, {"indicator": "ths_the_csrc_industry_stock", "indiparams": ["3", market_date]}, {"indicator": "ths_the_csrc_industry_code_stock", "indiparams": ["3", market_date]}, {"indicator": "ths_the_new_csrc_industry_stock", "indiparams": ["100", market_date]}, {"indicator": "ths_the_new_csrc_industry_code_stock", "indiparams": ["100", market_date]}, {"indicator": "ths_china_securities_industry_stock", "indiparams": ["1", market_date]}, {"indicator": "ths_china_securities_indus_code_stock", "indiparams": ["1", market_date]}, {"indicator": "ths_np_atoopc_stock", "indiparams": [report_date, "1"]}, {"indicator": "ths_operating_total_revenue_stock", "indiparams": [report_date, "100", "1"]}, {"indicator": "ths_total_equity_atoopc_stock", "indiparams": [report_date, "100", "1"]}, {"indicator": "ths_roe_stock", "indiparams": [report_date]}, {"indicator": "ths_pe_ttm_stock", "indiparams": [market_date, "100"]}, {"indicator": "ths_pb_mrq_stock", "indiparams": [market_date]}, {"indicator": "ths_turnover_ratio_y_stock", "indiparams": [market_date]}, {"indicator": "ths_fund_held_ratio_stock", "indiparams": [report_date]}, ], } def fetch_metrics( access_token: str, snapshot: SnapshotDate, stocks: list[dict[str, str]], raw_dir: Path, batch_size: int, refresh: bool, ) -> list[dict[str, Any]]: name_by_code = {item["code"]: item["name"] for item in stocks} output: list[dict[str, Any]] = [] code_list = list(name_by_code) for batch_no, batch in enumerate(chunks(code_list, batch_size), start=1): raw_path = raw_dir / f"metrics_v4_{snapshot.year}_size_{batch_size}_batch_{batch_no:03d}.json.gz" if raw_path.exists() and not refresh: with gzip.open(raw_path, "rt", encoding="utf-8") as handle: response = json.load(handle) else: response = post_json( "basic_data_service", metric_payload(batch, snapshot), {"access_token": access_token} ) write_raw(raw_path, response) for item in response.get("tables", []): code = str(item.get("thscode", "")) table = item.get("table", {}) row: dict[str, Any] = { "year": snapshot.year, "market_date": snapshot.market_date, "report_date": snapshot.report_date, "code": code, "name": name_by_code.get(code, ""), } for metric in METRICS: row[FIELD_LABELS[metric]] = first_value(table.get(metric)) output.append(row) print(f"{snapshot.year}: batch {batch_no}, received {len(response.get('tables', []))} stocks", flush=True) return output def fetch_industry_fallback( access_token: str, snapshot: SnapshotDate, rows: list[dict[str, Any]], raw_dir: Path, batch_size: int, refresh: bool, ) -> None: missing_codes = [row["code"] for row in rows if not str(row.get("csrc_industry") or "").strip()] if not missing_codes: return fallback: dict[str, tuple[Any, Any]] = {} for batch_no, batch in enumerate(chunks(missing_codes, batch_size), start=1): raw_path = raw_dir / f"industry_level1_fallback_{snapshot.year}_size_{batch_size}_batch_{batch_no:03d}.json.gz" if raw_path.exists() and not refresh: with gzip.open(raw_path, "rt", encoding="utf-8") as handle: response = json.load(handle) else: response = post_json( "basic_data_service", { "codes": ",".join(batch), "indipara": [ {"indicator": "ths_the_csrc_industry_stock", "indiparams": ["1", snapshot.market_date]}, {"indicator": "ths_the_csrc_industry_code_stock", "indiparams": ["1", snapshot.market_date]}, ], }, {"access_token": access_token}, ) write_raw(raw_path, response) for item in response.get("tables", []): table = item.get("table", {}) fallback[str(item.get("thscode", ""))] = ( first_value(table.get("ths_the_csrc_industry_stock")), first_value(table.get("ths_the_csrc_industry_code_stock")), ) for row in rows: if row["code"] in fallback: row["csrc_industry"], row["csrc_industry_code"] = fallback[row["code"]] def stable_industry(name: str) -> str: text = (name or "").strip() if not text: return "未映射" ordered_rules = ( ("医药卫生", ("医药", "医疗", "卫生", "生物制品")), ("金融地产", ("银行", "保险", "证券", "金融", "信托", "期货", "房地产")), ("公用事业", ("电力", "燃气生产", "煤气生产", "自来水", "水的生产", "公用事业", "公共设施")), ("通信服务", ("电信", "通信服务", "广播", "电视", "传媒", "出版", "新闻", "信息传播", "传播、文化", "艺术")), ("信息技术", ("计算机", "软件", "信息技术", "电子元器件", "电子器件", "半导体", "通信设备", "互联网")), ("能源", ("煤炭", "石油", "天然气", "油气", "炼焦", "采掘服务")), ("原材料", ("有色金属", "黑色金属", "金属矿", "非金属矿", "化学原料", "化学制品", "化工", "橡胶", "塑料", "造纸", "纸制品", "木材", "金属制品")), ("主要消费", ("农业", "林业", "畜牧", "渔业", "农、林、牧、渔", "食品", "饮料", "烟草", "农副食品")), ("可选消费", ("汽车", "家用电器", "零售", "批发", "商业经纪", "住宿", "旅馆", "餐饮", "旅游", "娱乐", "纺织", "服装", "家具", "教育", "其他社会服务")), ("工业", ("制造业", "设备", "机械", "仪器", "建筑", "装修装饰", "运输", "仓储", "邮政", "航空", "航天", "船舶", "铁路", "道路", "环境治理", "专业技术服务", "专业、科研服务", "租赁服务", "印刷")), ) for industry, keywords in ordered_rules: if any(keyword in text for keyword in keywords): return industry return "未映射" def stable_industry_for_row(row: dict[str, Any]) -> tuple[str, str]: csi = str(row.get("csi_industry_2021") or "").strip() csi_mapping = { "能源": "能源", "原材料": "原材料", "工业": "工业", "可选消费": "可选消费", "主要消费": "主要消费", "医药卫生": "医药卫生", "金融": "金融地产", "房地产": "金融地产", "信息技术": "信息技术", "通信服务": "通信服务", "公用事业": "公用事业", } if csi in csi_mapping: return csi_mapping[csi], "中证行业分类(2021)时点值" old_csrc = stable_industry(str(row.get("csrc_industry") or "")) if old_csrc != "未映射": return old_csrc, "证监会历史三级行业" new_code = str(row.get("new_csrc_industry_code") or "").strip() new_mapping = { "A": "主要消费", "D": "公用事业", "E": "工业", "F": "可选消费", "G": "工业", "H": "可选消费", "I": "信息技术", "J": "金融地产", "K": "金融地产", "L": "工业", "M": "工业", "N": "工业", "O": "可选消费", "P": "可选消费", "Q": "医药卫生", "R": "通信服务", } if new_code in new_mapping: return new_mapping[new_code], "新证监会门类时点值" return "未映射", "无可比细分行业" def as_number(value: Any) -> float | None: if value in (None, "", "--"): return None try: return float(value) except (TypeError, ValueError): return None def write_csv(path: Path, rows: list[dict[str, Any]], fieldnames: list[str]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8-sig", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() writer.writerows(rows) def write_company_snapshot(path: Path, rows: list[dict[str, Any]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) fields = [ "year", "market_date", "report_date", "code", "name", "csrc_industry", "csrc_industry_code", "new_csrc_industry", "new_csrc_industry_code", "csi_industry_2021", "csi_industry_2021_code", "stable_industry", "industry_mapping_source", "total_market_cap", "circulating_market_cap", "free_float_market_cap", "operating_revenue", "net_profit_parent", "equity_parent", "roe", "pe_ttm", "pb_mrq", "annual_turnover", "fund_held_ratio", ] with gzip.open(path, "wt", encoding="utf-8-sig", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fields, extrasaction="ignore") writer.writeheader() writer.writerows(rows) def median(values: list[float]) -> float | None: if not values: return None ordered = sorted(values) middle = len(ordered) // 2 if len(ordered) % 2: return ordered[middle] return (ordered[middle - 1] + ordered[middle]) / 2 def add_snapshot_percentiles(rows: list[dict[str, Any]], fields: tuple[str, ...]) -> None: for industry in {str(row["industry"]) for row in rows}: members = [row for row in rows if row["industry"] == industry] for field in fields: valid = [float(row[field]) for row in members if row.get(field) is not None] for row in members: value = row.get(field) if value is None or not valid: row[f"{field}_snapshot_percentile"] = None continue numeric = float(value) below = sum(item < numeric for item in valid) equal = sum(item == numeric for item in valid) row[f"{field}_snapshot_percentile"] = (below + equal / 2) / len(valid) def aggregate( rows: list[dict[str, Any]], ) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]: grouped: dict[tuple[int, str], list[dict[str, Any]]] = defaultdict(list) by_year: dict[int, list[dict[str, Any]]] = defaultdict(list) for row in rows: grouped[(int(row["year"]), str(row["stable_industry"]))].append(row) by_year[int(row["year"])].append(row) market_rows: list[dict[str, Any]] = [] profit_rows: list[dict[str, Any]] = [] metadata_rows: list[dict[str, Any]] = [] fundamental_rows: list[dict[str, Any]] = [] all_industries = (*STABLE_INDUSTRIES, "未映射") for year in sorted(by_year): year_rows = by_year[year] market_date = year_rows[0]["market_date"] report_date = year_rows[0]["report_date"] market_totals = { key: sum(value for row in year_rows if (value := as_number(row.get(key))) is not None) for key in ("total_market_cap", "circulating_market_cap", "free_float_market_cap") } total_profit = sum( value for row in year_rows if (value := as_number(row.get("net_profit_parent"))) is not None ) mapped_count = sum(row["stable_industry"] != "未映射" for row in year_rows) mapped_rows = [row for row in year_rows if row["stable_industry"] != "未映射"] mapped_market = { key: sum(value for row in mapped_rows if (value := as_number(row.get(key))) is not None) for key in ("total_market_cap", "circulating_market_cap", "free_float_market_cap") } metadata_rows.append( { "year": year, "market_date": market_date, "report_date": report_date, "stock_count": len(year_rows), "industry_mapped_count": mapped_count, "industry_mapping_rate": mapped_count / len(year_rows) if year_rows else None, "mapped_total_market_cap_rate": mapped_market["total_market_cap"] / market_totals["total_market_cap"] if market_totals["total_market_cap"] else None, "mapped_circulating_market_cap_rate": mapped_market["circulating_market_cap"] / market_totals["circulating_market_cap"] if market_totals["circulating_market_cap"] else None, "mapped_free_float_market_cap_rate": mapped_market["free_float_market_cap"] / market_totals["free_float_market_cap"] if market_totals["free_float_market_cap"] else None, "total_mv_valid_count": sum(as_number(row.get("total_market_cap")) is not None for row in year_rows), "free_float_mv_valid_count": sum(as_number(row.get("free_float_market_cap")) is not None for row in year_rows), "profit_valid_count": sum(as_number(row.get("net_profit_parent")) is not None for row in year_rows), "total_market_cap": market_totals["total_market_cap"], "total_net_profit_parent": total_profit, "source": "同花顺iFinD HTTP API", "calculated": "是", } ) for industry in all_industries: members = grouped.get((year, industry), []) values = { key: sum(value for row in members if (value := as_number(row.get(key))) is not None) for key in ("total_market_cap", "circulating_market_cap", "free_float_market_cap") } market_rows.append( { "year": year, "market_date": market_date, "industry": industry, "company_count": len(members), **values, "total_market_cap_share": values["total_market_cap"] / market_totals["total_market_cap"] if market_totals["total_market_cap"] else None, "circulating_market_cap_share": values["circulating_market_cap"] / market_totals["circulating_market_cap"] if market_totals["circulating_market_cap"] else None, "free_float_market_cap_share": values["free_float_market_cap"] / market_totals["free_float_market_cap"] if market_totals["free_float_market_cap"] else None, } ) industry_profit = sum( value for row in members if (value := as_number(row.get("net_profit_parent"))) is not None ) profit_rows.append( { "year": year, "report_date": report_date, "industry": industry, "company_count": len(members), "profit_valid_count": sum(as_number(row.get("net_profit_parent")) is not None for row in members), "net_profit_parent": industry_profit, "a_share_net_profit_parent": total_profit, "net_profit_share": industry_profit / total_profit if total_profit else None, } ) revenue_values = [ value for row in members if (value := as_number(row.get("operating_revenue"))) is not None ] equity_values = [ value for row in members if (value := as_number(row.get("equity_parent"))) is not None ] roe_values = [value for row in members if (value := as_number(row.get("roe"))) is not None] pe_values = [ value for row in members if (value := as_number(row.get("pe_ttm"))) is not None and value > 0 ] pb_values = [ value for row in members if (value := as_number(row.get("pb_mrq"))) is not None and value > 0 ] turnover_values = [ value for row in members if (value := as_number(row.get("annual_turnover"))) is not None ] fund_values = [ value for row in members if (value := as_number(row.get("fund_held_ratio"))) is not None ] revenue = sum(revenue_values) equity = sum(equity_values) fundamental_rows.append( { "year": year, "market_date": market_date, "report_date": report_date, "industry": industry, "company_count": len(members), "revenue_valid_count": len(revenue_values), "profit_valid_count": sum(as_number(row.get("net_profit_parent")) is not None for row in members), "equity_valid_count": len(equity_values), "roe_valid_count": len(roe_values), "pe_valid_count": len(pe_values), "pb_valid_count": len(pb_values), "turnover_valid_count": len(turnover_values), "fund_holding_valid_count": len(fund_values), "operating_revenue": revenue, "net_profit_parent": industry_profit, "equity_parent": equity, "aggregate_roe": industry_profit / equity if equity else None, "median_company_roe": median(roe_values), "median_company_pe_ttm_positive": median(pe_values), "median_company_pb_mrq_positive": median(pb_values), "median_company_annual_turnover": median(turnover_values), "median_company_fund_held_ratio": median(fund_values), } ) add_snapshot_percentiles( fundamental_rows, ("median_company_pe_ttm_positive", "median_company_pb_mrq_positive"), ) return market_rows, profit_rows, metadata_rows, fundamental_rows def mapping_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: counts: dict[tuple[str, str, str], int] = defaultdict(int) years: dict[tuple[str, str, str], set[int]] = defaultdict(set) for row in rows: key = (str(row.get("csrc_industry_code") or ""), str(row.get("csrc_industry") or ""), row["stable_industry"]) counts[key] += 1 years[key].add(int(row["year"])) return [ { "csrc_industry_code": key[0], "csrc_industry": key[1], "stable_industry": key[2], "company_year_observations": counts[key], "observed_years": ",".join(map(str, sorted(years[key]))), } for key in sorted(counts, key=lambda item: (item[2], item[0], item[1])) ] def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--years", nargs="*", type=int, default=[item.year for item in SNAPSHOTS]) parser.add_argument( "--annual", action="store_true", help="使用派生的1991-2025年末最后交易日表抓取连续年度快照", ) parser.add_argument("--batch-size", type=int, default=200) parser.add_argument("--refresh", action="store_true") return parser.parse_args() def annual_snapshots() -> tuple[SnapshotDate, ...]: path = Path(__file__).resolve().parents[1] / "data" / "research" / "derived" / "a_share_year_end_trading_dates_1991_2025.csv" if not path.exists(): raise SystemExit("Run scripts/fetch_market_year_end_dates.py first") with path.open(encoding="utf-8-sig") as handle: rows = list(csv.DictReader(handle)) return tuple( SnapshotDate( int(row["年份"]), str(row["年末最后交易日"])[:10], str(row["财务报告期末"])[:10], ) for row in rows ) def main() -> None: args = parse_args() refresh_token = os.environ.get("IFIND_REFRESH_TOKEN", "").strip() if not refresh_token: raise SystemExit("IFIND_REFRESH_TOKEN is required") access_token = exchange_access_token(refresh_token) root = Path(__file__).resolve().parents[1] raw_dir = root / "data/research/raw/ifind" derived_dir = root / "data/research/derived" available_snapshots = annual_snapshots() if args.annual else SNAPSHOTS selected = [item for item in available_snapshots if args.annual or item.year in set(args.years)] if not selected: raise SystemExit("No supported years selected") all_rows: list[dict[str, Any]] = [] for snapshot in selected: stocks = fetch_stock_pool(access_token, snapshot, raw_dir, args.refresh) print(f"{snapshot.year}: stock pool {len(stocks)}", flush=True) rows = fetch_metrics(access_token, snapshot, stocks, raw_dir, args.batch_size, args.refresh) fetch_industry_fallback(access_token, snapshot, rows, raw_dir, args.batch_size, args.refresh) for row in rows: row["stable_industry"], row["industry_mapping_source"] = stable_industry_for_row(row) all_rows.extend(rows) company_path = derived_dir / "a_share_company_snapshot_ifind.csv.gz" write_company_snapshot(company_path, all_rows) market_rows, profit_rows, metadata_rows, fundamental_rows = aggregate(all_rows) write_csv( derived_dir / "a_share_industry_market_cap_structure.csv", market_rows, ["year", "market_date", "industry", "company_count", "total_market_cap", "circulating_market_cap", "free_float_market_cap", "total_market_cap_share", "circulating_market_cap_share", "free_float_market_cap_share"], ) write_csv( derived_dir / "a_share_industry_profit_structure.csv", profit_rows, ["year", "report_date", "industry", "company_count", "profit_valid_count", "net_profit_parent", "a_share_net_profit_parent", "net_profit_share"], ) write_csv( derived_dir / "ifind_market_structure_metadata.csv", metadata_rows, ["year", "market_date", "report_date", "stock_count", "industry_mapped_count", "industry_mapping_rate", "mapped_total_market_cap_rate", "mapped_circulating_market_cap_rate", "mapped_free_float_market_cap_rate", "total_mv_valid_count", "free_float_mv_valid_count", "profit_valid_count", "total_market_cap", "total_net_profit_parent", "source", "calculated"], ) write_csv( derived_dir / "a_share_industry_fundamentals_valuation.csv", fundamental_rows, ["year", "market_date", "report_date", "industry", "company_count", "revenue_valid_count", "profit_valid_count", "equity_valid_count", "roe_valid_count", "pe_valid_count", "pb_valid_count", "turnover_valid_count", "fund_holding_valid_count", "operating_revenue", "net_profit_parent", "equity_parent", "aggregate_roe", "median_company_roe", "median_company_pe_ttm_positive", "median_company_pe_ttm_positive_snapshot_percentile", "median_company_pb_mrq_positive", "median_company_pb_mrq_positive_snapshot_percentile", "median_company_annual_turnover", "median_company_fund_held_ratio"], ) write_csv( derived_dir / "ifind_csrc_to_stable_industry_mapping.csv", mapping_rows(all_rows), ["csrc_industry_code", "csrc_industry", "stable_industry", "company_year_observations", "observed_years"], ) print(f"wrote {len(all_rows)} company-year rows to {company_path}") if __name__ == "__main__": main()