feat: 增加贵州茅台图表
This commit is contained in:
R524809
@@ -0,0 +1,405 @@
|
||||
"""Generate the chart-ready CSV layer for Guizhou Moutai from the two source workbooks.
|
||||
|
||||
The workbooks in data/moutai/rawData are retained unchanged. All values written by
|
||||
this script are in RMB 100m unless a CSV column name explicitly gives another unit.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
import pandas as pd
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
RAW = ROOT / "data" / "moutai" / "rawData"
|
||||
OUT = ROOT / "data" / "moutai"
|
||||
ANNUAL = RAW / "贵州茅台-年度.xlsx"
|
||||
QUARTERLY = RAW / "贵州茅台-季度.xlsx"
|
||||
|
||||
|
||||
def clean(value):
|
||||
if pd.isna(value):
|
||||
return None
|
||||
if isinstance(value, str):
|
||||
value = value.replace("\u200b", "").strip()
|
||||
return value or None
|
||||
return value
|
||||
|
||||
|
||||
def number(value):
|
||||
value = clean(value)
|
||||
if value is None:
|
||||
return None
|
||||
try:
|
||||
result = float(value)
|
||||
return result if math.isfinite(result) else None
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def label(value):
|
||||
value = clean(value)
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, float) and value.is_integer():
|
||||
return str(int(value))
|
||||
return str(value)
|
||||
|
||||
|
||||
def year(value):
|
||||
numeric = number(value)
|
||||
if numeric and 1900 <= numeric <= 2100:
|
||||
return str(int(numeric))
|
||||
match = re.search(r"(?:19|20)\d{2}", str(value or ""))
|
||||
return match.group(0) if match else None
|
||||
|
||||
|
||||
def quarter(value):
|
||||
text = str(value or "").replace(" ", "")
|
||||
match = re.search(r"(?:(20)?(\d{2}))Q([1-4])", text)
|
||||
if not match:
|
||||
return None
|
||||
return f"{match.group(1) or '20'}{match.group(2)}Q{match.group(3)}"
|
||||
|
||||
|
||||
def quarter_sort(period):
|
||||
text = quarter(period)
|
||||
return int(text[:4]) * 10 + int(text[-1])
|
||||
|
||||
|
||||
def sheet(workbook, name):
|
||||
return pd.read_excel(workbook, sheet_name=name, header=None).map(clean)
|
||||
|
||||
|
||||
def csv(name, rows, columns):
|
||||
OUT.mkdir(parents=True, exist_ok=True)
|
||||
frame = pd.DataFrame(rows, columns=columns)
|
||||
frame.to_csv(OUT / name, index=False, encoding="utf-8", float_format="%.6f")
|
||||
print(f"{name}: {len(frame)} rows")
|
||||
|
||||
|
||||
def wide_rows(frame, labels, start_col=1, header_row=0):
|
||||
periods = [(index, year(frame.iat[header_row, index])) for index in range(start_col, frame.shape[1])]
|
||||
periods = [(index, period) for index, period in periods if period]
|
||||
locations = {}
|
||||
for row_index in range(header_row + 1, frame.shape[0]):
|
||||
key = label(frame.iat[row_index, 0])
|
||||
if key in labels:
|
||||
locations[key] = row_index
|
||||
result = []
|
||||
for column, period in periods:
|
||||
row = {"period": period}
|
||||
for field, source_name in labels.items():
|
||||
source_row = locations.get(source_name)
|
||||
row[field] = number(frame.iat[source_row, column]) if source_row is not None else None
|
||||
result.append(row)
|
||||
return sorted(result, key=lambda row: int(row["period"]))
|
||||
|
||||
|
||||
def quarter_wide_rows(frame, labels, header_row=0):
|
||||
headers = [(index, label(frame.iat[header_row, index])) for index in range(1, frame.shape[1])]
|
||||
headers = [(index, quarter(value)) for index, value in headers]
|
||||
headers = [(index, value) for index, value in headers if value]
|
||||
locations = {}
|
||||
for row_index in range(header_row + 1, frame.shape[0]):
|
||||
key = label(frame.iat[row_index, 0])
|
||||
if key in labels:
|
||||
locations[key] = row_index
|
||||
result = []
|
||||
for column, period in headers:
|
||||
row = {"period": period}
|
||||
for field, source_name in labels.items():
|
||||
source_row = locations.get(source_name)
|
||||
row[field] = number(frame.iat[source_row, column]) if source_row is not None else None
|
||||
result.append(row)
|
||||
return sorted(result, key=lambda row: quarter_sort(row["period"]))
|
||||
|
||||
|
||||
def enrich_profit(rows):
|
||||
for index, row in enumerate(rows):
|
||||
prior = rows[index - 1] if index else None
|
||||
revenue = row.get("revenue")
|
||||
profit = row.get("net_profit")
|
||||
row["revenue_yoy"] = None
|
||||
row["net_profit_yoy"] = None
|
||||
if prior and revenue is not None and prior.get("revenue") not in (None, 0):
|
||||
row["revenue_yoy"] = revenue / prior["revenue"] - 1
|
||||
if prior and profit is not None and prior.get("net_profit") not in (None, 0):
|
||||
row["net_profit_yoy"] = profit / prior["net_profit"] - 1
|
||||
return rows
|
||||
|
||||
|
||||
def normalized_classified(frame):
|
||||
headers = [(index, label(frame.iat[1, index])) for index in range(2, frame.shape[1])]
|
||||
headers = [(index, quarter(value)) for index, value in headers]
|
||||
headers = [(index, value) for index, value in headers if value]
|
||||
wanted = {"茅台酒": "moutai", "系列酒": "series", "直销": "direct", "批发代理": "wholesale"}
|
||||
locations = {}
|
||||
for row_index in range(2, frame.shape[0]):
|
||||
key = label(frame.iat[row_index, 1])
|
||||
if key in wanted:
|
||||
locations[wanted[key]] = row_index
|
||||
rows = []
|
||||
for column, period in headers:
|
||||
row = {"period": period}
|
||||
for field in wanted.values():
|
||||
source_row = locations.get(field)
|
||||
row[field] = number(frame.iat[source_row, column]) if source_row is not None else None
|
||||
product_total = sum(value for value in (row["moutai"], row["series"]) if value is not None)
|
||||
channel_total = sum(value for value in (row["direct"], row["wholesale"]) if value is not None)
|
||||
row["moutai_share"] = row["moutai"] / product_total if product_total and row["moutai"] is not None else None
|
||||
row["direct_share"] = row["direct"] / channel_total if channel_total and row["direct"] is not None else None
|
||||
rows.append(row)
|
||||
return sorted(rows, key=lambda row: quarter_sort(row["period"]))
|
||||
|
||||
|
||||
def annual_product_or_channel(frame, source_names, header_row=1):
|
||||
headers = [(index, year(frame.iat[header_row, index])) for index in range(1, frame.shape[1])]
|
||||
headers = [(index, value) for index, value in headers if value]
|
||||
locations = {}
|
||||
for row_index in range(header_row + 1, frame.shape[0]):
|
||||
key = label(frame.iat[row_index, 0])
|
||||
if key in source_names:
|
||||
locations[source_names[key]] = row_index
|
||||
rows = []
|
||||
for column, period in headers:
|
||||
row = {"period": period}
|
||||
for field in source_names.values():
|
||||
source_row = locations.get(field)
|
||||
row[field] = number(frame.iat[source_row, column]) if source_row is not None else None
|
||||
rows.append(row)
|
||||
return sorted(rows, key=lambda row: int(row["period"]))
|
||||
|
||||
|
||||
def halfyear_composition(frame):
|
||||
headers = [(index, label(frame.iat[0, index])) for index in range(3, frame.shape[1])]
|
||||
headers = [(index, value) for index, value in headers if value and "H" in value]
|
||||
category = None
|
||||
locations = {}
|
||||
for row_index in range(1, frame.shape[0]):
|
||||
category = label(frame.iat[row_index, 1]) or category
|
||||
metric = label(frame.iat[row_index, 2])
|
||||
if category in {"茅台酒", "系列酒", "国内", "国外"} and metric == "营业收入":
|
||||
locations[category] = row_index
|
||||
key_map = {"茅台酒": "moutai", "系列酒": "series", "国内": "domestic", "国外": "overseas"}
|
||||
rows = []
|
||||
for column, period in headers:
|
||||
row = {"period": period}
|
||||
for category_name, field in key_map.items():
|
||||
source_row = locations.get(category_name)
|
||||
row[field] = number(frame.iat[source_row, column]) if source_row is not None else None
|
||||
rows.append(row)
|
||||
return sorted(rows, key=lambda row: (int(row["period"][:4]), row["period"][-1]))
|
||||
|
||||
|
||||
def cashflow_rows(frame, single_quarter=False):
|
||||
rows = []
|
||||
for row_index in range(1, frame.shape[0]):
|
||||
date = pd.to_datetime(frame.iat[row_index, 0], errors="coerce")
|
||||
if pd.isna(date):
|
||||
continue
|
||||
quarter = (date.month - 1) // 3 + 1
|
||||
rows.append({
|
||||
"period": f"{date.year}Q{quarter}" if single_quarter else str(date.year),
|
||||
"year": date.year,
|
||||
"quarter": quarter,
|
||||
"operating_cash_flow": number(frame.iat[row_index, 1]),
|
||||
"investing_cash_flow": number(frame.iat[row_index, 2]),
|
||||
"financing_cash_flow": number(frame.iat[row_index, 3]),
|
||||
"net_cash_increase": number(frame.iat[row_index, 4]),
|
||||
})
|
||||
if not single_quarter:
|
||||
return sorted(rows, key=lambda row: row["year"])
|
||||
cumulative = sorted(rows, key=lambda row: (row["year"], row["quarter"]))
|
||||
previous = {}
|
||||
for row in cumulative:
|
||||
prior = previous.get(row["year"])
|
||||
if prior:
|
||||
for field in ("operating_cash_flow", "investing_cash_flow", "financing_cash_flow", "net_cash_increase"):
|
||||
row[field] = row[field] - prior[field] if row[field] is not None and prior[field] is not None else None
|
||||
previous[row["year"]] = {field: row[field] if prior is None else (row[field] + prior[field] if row[field] is not None and prior[field] is not None else None) for field in ("operating_cash_flow", "investing_cash_flow", "financing_cash_flow", "net_cash_increase")}
|
||||
return cumulative
|
||||
|
||||
|
||||
def annual_dividend(frame):
|
||||
rows = []
|
||||
for row_index in range(1, frame.shape[0]):
|
||||
period = year(frame.iat[row_index, 0])
|
||||
if not period:
|
||||
continue
|
||||
dividend = number(frame.iat[row_index, 4])
|
||||
profit = number(frame.iat[row_index, 6])
|
||||
if dividend is None and profit is None:
|
||||
continue
|
||||
rows.append({
|
||||
"period": period,
|
||||
"dividend": dividend,
|
||||
"net_profit": profit,
|
||||
"payout_ratio": dividend / profit if dividend is not None and profit not in (None, 0) else None,
|
||||
})
|
||||
return sorted(rows, key=lambda row: int(row["period"]))
|
||||
|
||||
|
||||
def price_adjustments(frame):
|
||||
rows = []
|
||||
for row_index in range(1, frame.shape[0]):
|
||||
date = pd.to_datetime(frame.iat[row_index, 0], errors="coerce")
|
||||
before = number(frame.iat[row_index, 1])
|
||||
after = number(frame.iat[row_index, 2])
|
||||
if pd.isna(date) or after is None:
|
||||
continue
|
||||
rows.append({
|
||||
"period": date.strftime("%Y-%m"),
|
||||
"before_price": before,
|
||||
"after_price": after,
|
||||
"increase_rate": number(frame.iat[row_index, 4]),
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
def ton_price(frame):
|
||||
headers = [(column, year(frame.iat[1, column])) for column in range(1, frame.shape[1])]
|
||||
headers = [(column, period) for column, period in headers if period]
|
||||
names = {"茅台酒营收": "moutai_revenue", "系列酒营收": "series_revenue", "茅台商品酒销量（吨）": "moutai_volume", "系列酒销量（吨）": "series_volume"}
|
||||
locations = {}
|
||||
for row_index in range(2, frame.shape[0]):
|
||||
source = label(frame.iat[row_index, 0])
|
||||
if source in names:
|
||||
locations[names[source]] = row_index
|
||||
rows = []
|
||||
for column, period in headers:
|
||||
row = {"period": period}
|
||||
for field in names.values():
|
||||
source_row = locations.get(field)
|
||||
row[field] = number(frame.iat[source_row, column]) if source_row is not None else None
|
||||
row["moutai_price_per_ton"] = row["moutai_revenue"] * 10000 / row["moutai_volume"] if row["moutai_revenue"] is not None and row["moutai_volume"] not in (None, 0) else None
|
||||
row["series_price_per_ton"] = row["series_revenue"] * 10000 / row["series_volume"] if row["series_revenue"] is not None and row["series_volume"] not in (None, 0) else None
|
||||
rows.append(row)
|
||||
return sorted(rows, key=lambda row: int(row["period"]))
|
||||
|
||||
|
||||
def annual_profitability(frame):
|
||||
headers = {label(frame.iat[0, column]): column for column in range(1, frame.shape[1])}
|
||||
field_names = {"营业收入": "revenue", "毛利": "gross_profit", "毛利率": "gross_margin", "归母净利润": "net_profit"}
|
||||
rows = []
|
||||
for row_index in range(1, frame.shape[0]):
|
||||
period = year(frame.iat[row_index, 0])
|
||||
if not period:
|
||||
continue
|
||||
row = {"period": period}
|
||||
for source, field in field_names.items():
|
||||
column = headers.get(source)
|
||||
row[field] = number(frame.iat[row_index, column]) if column is not None else None
|
||||
if any(row[field] is not None for field in field_names.values()):
|
||||
rows.append(row)
|
||||
return sorted(rows, key=lambda row: int(row["period"]))
|
||||
|
||||
|
||||
def distributors(frame):
|
||||
rows = []
|
||||
for row_index in range(3, frame.shape[0]):
|
||||
period = label(frame.iat[row_index, 0])
|
||||
cutoff = pd.to_datetime(frame.iat[row_index, 1], errors="coerce")
|
||||
if not period or pd.isna(cutoff):
|
||||
continue
|
||||
domestic = number(frame.iat[row_index, 2])
|
||||
overseas = number(frame.iat[row_index, 5])
|
||||
total = number(frame.iat[row_index, 8])
|
||||
if total is None:
|
||||
continue
|
||||
rows.append({"period": period.replace(" ", ""), "domestic": domestic, "overseas": overseas, "total": total})
|
||||
return rows
|
||||
|
||||
|
||||
def main():
|
||||
annual_profit = enrich_profit(annual_profitability(sheet(ANNUAL, "盈利能力")))
|
||||
csv("moutai_profitability_annual.csv", annual_profit, ["period", "revenue", "gross_profit", "net_profit", "gross_margin", "revenue_yoy", "net_profit_yoy"])
|
||||
|
||||
quarterly_profit = sheet(QUARTERLY, "盈利能力")
|
||||
field_names = {"营业收入": "revenue", "毛利": "gross_profit", "毛利率": "gross_margin", "归母净利润": "net_profit"}
|
||||
metric_columns = {field: next((column for column in range(8, quarterly_profit.shape[1]) if label(quarterly_profit.iat[1, column]) == source), None) for source, field in field_names.items()}
|
||||
quarterly_rows = []
|
||||
for row_index in range(2, quarterly_profit.shape[0]):
|
||||
period = quarter(quarterly_profit.iat[row_index, 8])
|
||||
if not period:
|
||||
continue
|
||||
row = {"period": period}
|
||||
for field, column in metric_columns.items():
|
||||
row[field] = number(quarterly_profit.iat[row_index, column]) if column is not None else None
|
||||
quarterly_rows.append(row)
|
||||
quarterly_rows = enrich_profit(sorted(quarterly_rows, key=lambda row: quarter_sort(row["period"])))
|
||||
csv("moutai_profitability_quarterly.csv", quarterly_rows, ["period", "revenue", "gross_profit", "net_profit", "gross_margin", "revenue_yoy", "net_profit_yoy"])
|
||||
|
||||
annual_expense = wide_rows(sheet(ANNUAL, "费用率"), {
|
||||
"sales_expense": "销售费用", "admin_expense": "管理费用", "rd_expense": "研发费用", "tax_surcharge": "税金及附加", "revenue": "营业收入",
|
||||
})
|
||||
for row in annual_expense:
|
||||
for expense in ("sales_expense", "admin_expense", "rd_expense", "tax_surcharge"):
|
||||
row[f"{expense}_rate"] = row[expense] / row["revenue"] if row[expense] is not None and row["revenue"] not in (None, 0) else None
|
||||
csv("moutai_expense_annual.csv", annual_expense, ["period", "sales_expense", "admin_expense", "rd_expense", "tax_surcharge", "revenue", "sales_expense_rate", "admin_expense_rate", "rd_expense_rate", "tax_surcharge_rate"])
|
||||
|
||||
quarter_expense = quarter_wide_rows(sheet(QUARTERLY, "费用率"), {
|
||||
"sales_expense": "销售费用", "admin_expense": "管理费用", "rd_expense": "研发费用", "tax_surcharge": "税金及附加", "revenue": "营业收入", "gross_margin": "毛利率",
|
||||
}, header_row=1)
|
||||
for row in quarter_expense:
|
||||
for expense in ("sales_expense", "admin_expense", "rd_expense", "tax_surcharge"):
|
||||
row[f"{expense}_rate"] = row[expense] / row["revenue"] if row[expense] is not None and row["revenue"] not in (None, 0) else None
|
||||
csv("moutai_expense_quarterly.csv", quarter_expense, ["period", "sales_expense", "admin_expense", "rd_expense", "tax_surcharge", "revenue", "gross_margin", "sales_expense_rate", "admin_expense_rate", "rd_expense_rate", "tax_surcharge_rate"])
|
||||
|
||||
balance_fields = {"total_assets": "资产总计", "equity": "所有者权益合计", "liabilities": "负债合计", "cash_like": "类现金", "cash": "货币资金", "receivables": "应收账款", "inventory": "存货"}
|
||||
csv("moutai_balance_annual.csv", wide_rows(sheet(ANNUAL, "资产负债表"), balance_fields), ["period", *balance_fields])
|
||||
csv("moutai_balance_quarterly.csv", quarter_wide_rows(sheet(QUARTERLY, "资产负债表"), balance_fields), ["period", *balance_fields])
|
||||
|
||||
cf_columns = ["period", "operating_cash_flow", "investing_cash_flow", "financing_cash_flow", "net_cash_increase"]
|
||||
csv("moutai_cashflow_annual.csv", cashflow_rows(sheet(ANNUAL, "现金流量表")), cf_columns)
|
||||
csv("moutai_cashflow_quarterly.csv", cashflow_rows(sheet(QUARTERLY, "现金流量表"), single_quarter=True), cf_columns)
|
||||
|
||||
product = annual_product_or_channel(sheet(ANNUAL, "酒类占比"), {"茅台酒营收": "moutai", "系列酒营收": "series"}, header_row=0)
|
||||
for row in product:
|
||||
total = sum(value for value in (row["moutai"], row["series"]) if value is not None)
|
||||
row["moutai_share"] = row["moutai"] / total if total and row["moutai"] is not None else None
|
||||
row["series_share"] = row["series"] / total if total and row["series"] is not None else None
|
||||
csv("moutai_product_revenue_annual.csv", product, ["period", "moutai", "series", "moutai_share", "series_share"])
|
||||
|
||||
channels = annual_product_or_channel(sheet(ANNUAL, "渠道占比"), {"直销营收（亿）": "direct", "批发代理营收（亿）": "wholesale"})
|
||||
for row in channels:
|
||||
total = sum(value for value in (row["direct"], row["wholesale"]) if value is not None)
|
||||
row["direct_share"] = row["direct"] / total if total and row["direct"] is not None else None
|
||||
row["wholesale_share"] = row["wholesale"] / total if total and row["wholesale"] is not None else None
|
||||
csv("moutai_channel_revenue_annual.csv", channels, ["period", "direct", "wholesale", "direct_share", "wholesale_share"])
|
||||
|
||||
classified_columns = ["period", "moutai", "series", "direct", "wholesale", "moutai_share", "direct_share"]
|
||||
csv("moutai_classified_revenue_quarterly.csv", normalized_classified(sheet(QUARTERLY, "分类营收-单机")), classified_columns)
|
||||
csv("moutai_classified_revenue_cumulative.csv", normalized_classified(sheet(QUARTERLY, "分类营收-累计")), classified_columns)
|
||||
csv("moutai_revenue_composition_halfyear.csv", halfyear_composition(sheet(QUARTERLY, "营收构成-累计")), ["period", "moutai", "series", "domestic", "overseas"])
|
||||
|
||||
liquor = sheet(ANNUAL, "全国白酒产量")
|
||||
liquor_rows = [{"period": year(liquor.iat[i, 0]), "output": number(liquor.iat[i, 1]), "yoy": number(liquor.iat[i, 2])} for i in range(1, liquor.shape[0])]
|
||||
liquor_rows = [row for row in liquor_rows if row["period"]]
|
||||
csv("moutai_liquor_industry_output.csv", liquor_rows, ["period", "output", "yoy"])
|
||||
|
||||
base = sheet(ANNUAL, "茅台基酒产销量")
|
||||
production = {}
|
||||
sales = {}
|
||||
for i in range(2, base.shape[0]):
|
||||
production[year(base.iat[i, 0])] = {"planned_capacity": number(base.iat[i, 1]), "released_capacity": number(base.iat[i, 2]), "base_wine_output": number(base.iat[i, 3])}
|
||||
sales[year(base.iat[i, 6])] = number(base.iat[i, 7])
|
||||
volume_rows = []
|
||||
for period in sorted({value for value in production if value} | {value for value in sales if value}, key=int):
|
||||
volume_rows.append({"period": period, **production.get(period, {}), "commercial_wine_sales": sales.get(period)})
|
||||
csv("moutai_base_wine_volume.csv", volume_rows, ["period", "planned_capacity", "released_capacity", "base_wine_output", "commercial_wine_sales"])
|
||||
|
||||
price = sheet(ANNUAL, "飞天茅台历年批价")
|
||||
price_rows = [{"period": year(price.iat[i, 0]), "wholesale_price": number(price.iat[i, 1])} for i in range(1, price.shape[0])]
|
||||
csv("moutai_flying_price.csv", [row for row in price_rows if row["period"] and row["wholesale_price"] is not None], ["period", "wholesale_price"])
|
||||
csv("moutai_price_adjustments.csv", price_adjustments(sheet(ANNUAL, "提价记录")), ["period", "before_price", "after_price", "increase_rate"])
|
||||
csv("moutai_ton_price.csv", ton_price(sheet(ANNUAL, "吨价")), ["period", "moutai_revenue", "series_revenue", "moutai_volume", "series_volume", "moutai_price_per_ton", "series_price_per_ton"])
|
||||
csv("moutai_dividend.csv", annual_dividend(sheet(ANNUAL, "历年分红")), ["period", "dividend", "net_profit", "payout_ratio"])
|
||||
csv("moutai_distributors.csv", distributors(sheet(QUARTERLY, "经销商数量")), ["period", "domestic", "overseas", "total"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user