【最新2025】华证 ESG 评级评分数据库：2009-2025 上市公司季度/年度多维面板数据包与清洗入库指南

import polars as pl
from typing import Dict

# 华证 9 档评级标准数值映射表
RATING_MAP: Dict[str, int] = {
    "AAA": 9, "AA": 8, "A": 7,
    "BBB": 6, "BB": 5, "B": 4,
    "CCC": 3, "CC": 2, "C": 1
}

def clean_esg_dataset(file_path: str, is_quarterly: bool = False) -> pl.DataFrame:
    # 1. 高速读取原始数据包
    df = pl.read_csv(file_path, infer_schema_length=10000)

    # 2. 规范化字段命名
    cleaned = df.rename({
        "证券代码": "symbol_raw",
        "年份": "report_year",
        "证券简称": "company_name",
        "综合评级": "overall_rating",
        "综合赋值": "overall_numeric_rank",
        "综合得分": "overall_score",
        "E评级": "e_rating",
        "E得分": "e_score",
        "S评级": "s_rating",
        "S得分": "s_score",
        "G评级": "g_rating",
        "G得分": "g_score",
        "行业": "industry"
    })

    # 3. 代码标准化：补齐 6 位前导零并添加交易所后缀
    cleaned = cleaned.with_columns([
        pl.col("symbol_raw").cast(pl.Utf8).str.zfill(6).alias("symbol"),
        pl.col("report_year").cast(pl.Int32)
    ])

    cleaned = cleaned.with_columns(
        pl.when(pl.col("symbol").str.starts_with("6"))
          .then(pl.lit("SH"))
          .when(pl.col("symbol").str.starts_with("0") | pl.col("symbol").str.starts_with("3"))
          .then(pl.lit("SZ"))
          .when(pl.col("symbol").str.starts_with("8") | pl.col("symbol").str.starts_with("4") | pl.col("symbol").str.starts_with("9"))
          .then(pl.lit("BJ"))
          .otherwise(pl.lit("UNKNOWN"))
          .alias("exchange")
    )

    # 4. 数值类型转换与映射校验
    cleaned = cleaned.with_columns([
        pl.col("overall_rating").map_elements(lambda r: RATING_MAP.get(str(r).strip(), None), return_dtype=pl.Int32).alias("computed_rank"),
        pl.col("overall_score").cast(pl.Float64),
        pl.col("e_score").cast(pl.Float64),
        pl.col("s_score").cast(pl.Float64),
        pl.col("g_score").cast(pl.Float64),
    ])

    # 5. 特征工程：计算 ESG 三支柱离散度（衡量企业表现均衡度）
    cleaned = cleaned.with_columns(
        (
            ((pl.col("e_score") - pl.col("overall_score")).pow(2) +
             (pl.col("s_score") - pl.col("overall_score")).pow(2) +
             (pl.col("g_score") - pl.col("overall_score")).pow(2)) / 3.0
        ).sqrt().alias("esg_pillar_dispersion")
    )

    return cleaned

if __name__ == "__main__":
    quarterly_df = clean_esg_dataset("huazheng_esg_quarterly_2009_2025.csv", is_quarterly=True)
    print(f"✅ 华证 ESG 季度数据包加载成功，总记录数: {len(quarterly_df)}")

-- 创建华证 ESG 时序面板表
CREATE TABLE listed_company_esg_ratings (
    id BIGSERIAL,
    symbol VARCHAR(12) NOT NULL,            -- 股票代码如 000001.SZ
    company_name VARCHAR(64) NOT NULL,
    period_type VARCHAR(8) NOT NULL,        -- 'ANNUAL' 或 'QUARTER'
    period_date DATE NOT NULL,              -- 报告期结束日期
    report_year INT NOT NULL,
    industry VARCHAR(64),

    overall_rating VARCHAR(4) NOT NULL,
    overall_rank INT NOT NULL CHECK (overall_rank BETWEEN 1 AND 9),
    overall_score NUMERIC(5, 2) NOT NULL,

    e_rating VARCHAR(4),
    e_score NUMERIC(5, 2),
    s_rating VARCHAR(4),
    s_score NUMERIC(5, 2),
    g_rating VARCHAR(4),
    g_score NUMERIC(5, 2),

    pillar_dispersion NUMERIC(5, 2),
    created_at TIMESTAMPTZ DEFAULT NOW(),

    PRIMARY KEY (symbol, period_type, period_date)
) PARTITION BY RANGE (period_date);

-- 创建 2020-2025 分区
CREATE TABLE esg_ratings_2020_2025 PARTITION OF listed_company_esg_ratings
    FOR VALUES FROM ('2020-01-01') TO ('2026-01-01');

-- 建立回测常用复合索引
CREATE INDEX idx_esg_industry_period ON listed_company_esg_ratings (industry, period_date, overall_rank DESC);
CREATE INDEX idx_esg_symbol_timeline ON listed_company_esg_ratings (symbol, period_date DESC);