"""策略2：绘制中证 A500 与沪深 300 成立以来双轴收盘价。""" from __future__ import annotations import argparse from dataclasses import dataclass from datetime import date from pathlib import Path from typing import Dict, Iterable, List, Sequence, Tuple from xml.sax.saxutils import escape from sidecar.config import load_config from sidecar.models import KlineBar from sidecar.service import UnifiedDataService from sidecar.sources.tencent import TencentFinanceSource from sidecar.storage import DuckDBStore from strategy.a500_close_chart import A500_START_DATE, A500_SYMBOL, ClosePoint, filter_close_points HS300_SYMBOL = "sh000300" DEFAULT_OUTPUT_PATH = Path("strategy/output/a500_hs300_close.svg") @dataclass(frozen=True) class DualAxisSeries: """双轴图数据序列。""" name: str symbol: str color: str points: List[ClosePoint] def align_points_by_date(left_points: Sequence[ClosePoint], right_points: Sequence[ClosePoint]) -> Tuple[List[ClosePoint], List[ClosePoint]]: """ 功能说明：按共同交易日对齐两组收盘价点。 参数说明：left_points 为左轴序列，right_points 为右轴序列。 返回值说明：返回按日期升序排列的两组 ClosePoint。 注意事项：只保留双方同时存在的交易日，避免横轴错位。 """ left_map = {point.trade_date: point for point in left_points} right_map = {point.trade_date: point for point in right_points} dates = sorted(set(left_map).intersection(right_map)) return [left_map[item] for item in dates], [right_map[item] for item in dates] def build_dual_axis_svg_chart(left: DualAxisSeries, right: DualAxisSeries, title: str, width: int = 1200, height: int = 680) -> str: """ 功能说明：把两组收盘价序列渲染为双 Y 轴 SVG 折线图。 参数说明：left 为左轴序列，right 为右轴序列，title 为标题，width 和 height 为画布尺寸。 返回值说明：返回完整 SVG 文本。 注意事项：两组序列按首日价格归一化，共享涨跌比例坐标，确保相同涨跌比例对应相同纵向距离。 """ if not left.points or not right.points: raise ValueError("没有可绘制的双轴收盘价数据") if len(left.points) != len(right.points): raise ValueError("双轴序列长度不一致，请先按交易日对齐") margin_left = 82 margin_right = 82 margin_top = 78 margin_bottom = 86 chart_width = width - margin_left - margin_right chart_height = height - margin_top - margin_bottom left_base = left.points[0].close right_base = right.points[0].close ratio_min, ratio_max = _shared_ratio_range(left.points, right.points) ratio_range = ratio_max - ratio_min or 1.0 left_coordinates = _scale_points_by_ratio(left.points, left_base, margin_left, margin_top, chart_width, chart_height, ratio_min, ratio_range) right_coordinates = _scale_points_by_ratio(right.points, right_base, margin_left, margin_top, chart_width, chart_height, ratio_min, ratio_range) ratio_ticks = _build_y_ticks(ratio_min, ratio_max, 5) y_axis = _render_left_axis_by_ratio(ratio_ticks, left_base, margin_left, margin_top, chart_width, chart_height, ratio_min, ratio_range, left.color) y_axis += _render_right_axis_by_ratio(ratio_ticks, right_base, width - margin_right, margin_top, chart_height, ratio_min, ratio_range, right.color) x_axis = _render_x_axis(_build_x_ticks(left.points, 6), left_coordinates, margin_top + chart_height) first_date = left.points[0].trade_date last_date = left.points[-1].trade_date subtitle = "%s 至 %s，共 %d 个共同交易日；纵轴按首日收盘价归一化" % (first_date.isoformat(), last_date.isoformat(), len(left.points)) return """ {title} {subtitle} {left_name}（左轴） {right_name}（右轴） {y_axis} {x_axis} 数据来源：sidecar / 腾讯财经；双轴按首日价格归一化，等距离代表等涨跌比例 """.format( width=width, height=height, center_x=width / 2, title=escape(title), subtitle=escape(subtitle), legend_left_x=width / 2 - 140, legend_left_text_x=width / 2 - 128, legend_right_x=width / 2 + 40, legend_right_text_x=width / 2 + 52, left_color=left.color, right_color=right.color, left_name=escape(left.name), right_name=escape(right.name), y_axis="\n ".join(y_axis), x_axis="\n ".join(x_axis), left_axis=margin_left, right_axis=width - margin_right, top=margin_top, bottom=margin_top + chart_height, left_polyline=" ".join("%.2f,%.2f" % coordinate for coordinate in left_coordinates), right_polyline=" ".join("%.2f,%.2f" % coordinate for coordinate in right_coordinates), footer_y=height - 18, ) def fetch_dual_axis_series(refresh: bool = False) -> Tuple[DualAxisSeries, DualAxisSeries]: """ 功能说明：通过 sidecar 获取 A500 和沪深 300 的对齐收盘价序列。 参数说明：refresh 表示是否强制刷新 sidecar 缓存。 返回值说明：返回 A500 左轴序列和沪深 300 右轴序列。 注意事项：横轴范围以 A500 成立日为起点，并只保留共同交易日。 """ config = load_config() tencent = TencentFinanceSource(config.request_timeout_seconds) service = UnifiedDataService(DuckDBStore(config.db_path), config, tencent, [tencent]) a500_points = filter_close_points(service.get_kline(A500_SYMBOL, period="day", limit=800, refresh=refresh), A500_START_DATE) hs300_points = filter_close_points(service.get_kline(HS300_SYMBOL, period="day", limit=800, refresh=refresh), A500_START_DATE) aligned_a500, aligned_hs300 = align_points_by_date(a500_points, hs300_points) return ( DualAxisSeries("中证 A500", A500_SYMBOL, "#2563eb", aligned_a500), DualAxisSeries("沪深 300", HS300_SYMBOL, "#dc2626", aligned_hs300), ) def write_a500_hs300_close_chart(output_path: Path = DEFAULT_OUTPUT_PATH, refresh: bool = False) -> Path: """ 功能说明：生成 A500 与沪深 300 双轴收盘价 SVG 图。 参数说明：output_path 为输出 SVG 路径，refresh 表示是否强制刷新 sidecar 缓存。 返回值说明：返回实际写入的图片路径。 注意事项：输出目录不存在时会自动创建。 """ left, right = fetch_dual_axis_series(refresh=refresh) svg = build_dual_axis_svg_chart(left, right, "中证 A500 与沪深 300 收盘价对比") output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text(svg, encoding="utf-8") return output_path def parse_args() -> argparse.Namespace: """ 功能说明：解析命令行参数。 参数说明：无。 返回值说明：返回 argparse.Namespace 参数对象。 注意事项：默认输出到 strategy/output/a500_hs300_close.svg。 """ parser = argparse.ArgumentParser(description="绘制中证 A500 与沪深 300 双轴收盘价") parser.add_argument("--output", default=str(DEFAULT_OUTPUT_PATH), help="输出 SVG 文件路径") parser.add_argument("--refresh", action="store_true", help="强制刷新 sidecar 缓存") return parser.parse_args() def main() -> None: """ 功能说明：命令行入口函数。 参数说明：无。 返回值说明：无返回值。 注意事项：执行成功后会在终端打印输出文件路径。 """ args = parse_args() output_path = write_a500_hs300_close_chart(Path(args.output), refresh=args.refresh) print("已生成图表: %s" % output_path) def _shared_ratio_range(left_points: Sequence[ClosePoint], right_points: Sequence[ClosePoint]) -> Tuple[float, float]: """ 功能说明：计算两组序列共享的涨跌比例范围。 参数说明：left_points 为左轴序列，right_points 为右轴序列。 返回值说明：返回最小涨跌比例和最大涨跌比例，例如 0.1 表示上涨 10%。 注意事项：涨跌比例都以各自首个收盘价为基准。 """ ratios = _ratios_from_base(left_points) + _ratios_from_base(right_points) return min(ratios), max(ratios) def _ratios_from_base(points: Sequence[ClosePoint]) -> List[float]: """ 功能说明：计算序列相对首个收盘价的涨跌比例。 参数说明：points 为收盘价点序列。 返回值说明：返回涨跌比例列表。 注意事项：首个点的涨跌比例固定为 0。 """ base = points[0].close if base == 0: raise ValueError("基准收盘价不能为 0") return [(point.close / base) - 1 for point in points] def _scale_points_by_ratio(points: Sequence[ClosePoint], base_close: float, margin_left: int, margin_top: int, chart_width: int, chart_height: int, ratio_min: float, ratio_range: float) -> List[Tuple[float, float]]: """ 功能说明：按涨跌比例把收盘价点转换为 SVG 坐标。 参数说明：points 为收盘价点，base_close 为首日收盘价，margin_left 和 margin_top 为边距，chart_width 和 chart_height 为绘图区尺寸，ratio_min 和 ratio_range 为共享比例缩放参数。 返回值说明：返回 SVG 坐标列表。 注意事项：不同指数相同涨跌比例会得到相同 Y 坐标。 """ if base_close == 0: raise ValueError("基准收盘价不能为 0") max_index = max(len(points) - 1, 1) coordinates = [] for index, point in enumerate(points): ratio = point.close / base_close - 1 x_position = margin_left + index / max_index * chart_width y_position = margin_top + chart_height - ((ratio - ratio_min) / ratio_range) * chart_height coordinates.append((x_position, y_position)) return coordinates def _price_from_ratio(base_close: float, ratio: float) -> float: """ 功能说明：把涨跌比例转换为某个指数自身价格。 参数说明：base_close 为指数首日收盘价，ratio 为涨跌比例。 返回值说明：返回该比例对应的价格。 注意事项：用于左右 Y 轴在同一比例位置显示各自价格刻度。 """ return base_close * (1 + ratio) def _render_left_axis_by_ratio(ticks: Sequence[float], base_close: float, margin_left: int, margin_top: int, chart_width: int, chart_height: int, ratio_min: float, ratio_range: float, color: str) -> List[str]: """ 功能说明：按共享涨跌比例渲染左侧价格轴。 参数说明：ticks 为比例刻度，base_close 为左轴基准价，margin_left 和 margin_top 为边距，chart_width 和 chart_height 为绘图区尺寸，ratio_min 和 ratio_range 为比例缩放参数，color 为文本颜色。 返回值说明：返回 SVG 片段列表。 注意事项：左轴显示 A500 价格，但刻度位置由共享涨跌比例决定。 """ lines = [] for tick_ratio in ticks: y_position = margin_top + chart_height - ((tick_ratio - ratio_min) / ratio_range) * chart_height price = _price_from_ratio(base_close, tick_ratio) percent = tick_ratio * 100 lines.append('' % (margin_left, y_position, margin_left + chart_width, y_position)) lines.append('%.2f (%+.1f%%)' % (margin_left - 10, y_position + 4, color, price, percent)) return lines def _render_right_axis_by_ratio(ticks: Sequence[float], base_close: float, right_axis: int, margin_top: int, chart_height: int, ratio_min: float, ratio_range: float, color: str) -> List[str]: """ 功能说明：按共享涨跌比例渲染右侧价格轴。 参数说明：ticks 为比例刻度，base_close 为右轴基准价，right_axis 为右轴横坐标，margin_top 和 chart_height 为绘图区参数，ratio_min 和 ratio_range 为比例缩放参数，color 为文本颜色。 返回值说明：返回 SVG 片段列表。 注意事项：右轴显示沪深 300 价格，但刻度位置由共享涨跌比例决定。 """ lines = [] for tick_ratio in ticks: y_position = margin_top + chart_height - ((tick_ratio - ratio_min) / ratio_range) * chart_height price = _price_from_ratio(base_close, tick_ratio) percent = tick_ratio * 100 lines.append('%.2f (%+.1f%%)' % (right_axis + 10, y_position + 4, color, price, percent)) return lines def _value_range(points: Sequence[ClosePoint]) -> Tuple[float, float]: """ 功能说明：计算收盘价范围。 参数说明：points 为收盘价点序列。 返回值说明：返回最低值和最高值。 注意事项：调用方需保证 points 非空。 """ values = [point.close for point in points] return min(values), max(values) def _scale_points(points: Sequence[ClosePoint], margin_left: int, margin_top: int, chart_width: int, chart_height: int, min_close: float, close_range: float) -> List[Tuple[float, float]]: """ 功能说明：把收盘价点转换为 SVG 坐标。 参数说明：points 为收盘价点，margin_left 和 margin_top 为边距，chart_width 和 chart_height 为绘图区尺寸，min_close 和 close_range 为价格缩放参数。 返回值说明：返回 SVG 坐标列表。 注意事项：单点数据会绘制在横轴起点。 """ max_index = max(len(points) - 1, 1) return [(margin_left + index / max_index * chart_width, margin_top + chart_height - ((point.close - min_close) / close_range) * chart_height) for index, point in enumerate(points)] def _build_y_ticks(min_close: float, max_close: float, tick_count: int) -> List[float]: """ 功能说明：生成纵轴价格刻度。 参数说明：min_close 为最低收盘价，max_close 为最高收盘价，tick_count 为刻度数量。 返回值说明：返回价格刻度列表。 注意事项：当最高最低相等时返回单一价格刻度。 """ if tick_count <= 1 or min_close == max_close: return [min_close] step = (max_close - min_close) / (tick_count - 1) return [min_close + step * index for index in range(tick_count)] def _build_x_ticks(points: Sequence[ClosePoint], tick_count: int) -> List[Tuple[int, date]]: """ 功能说明：生成横轴日期刻度。 参数说明：points 为收盘价点序列，tick_count 为目标刻度数量。 返回值说明：返回二元组列表，包含点索引和日期。 注意事项：会去重索引，避免短序列重复显示同一天。 """ if not points: return [] max_index = len(points) - 1 if tick_count <= 1 or max_index == 0: return [(0, points[0].trade_date)] indexes = sorted({round(index * max_index / (tick_count - 1)) for index in range(tick_count)}) return [(index, points[index].trade_date) for index in indexes] def _render_left_axis(ticks: Sequence[float], margin_left: int, margin_top: int, chart_width: int, chart_height: int, min_close: float, close_range: float, color: str) -> List[str]: """ 功能说明：渲染左侧纵轴网格和刻度。 参数说明：ticks 为价格刻度，margin_left 和 margin_top 为边距，chart_width 和 chart_height 为绘图区尺寸，min_close 和 close_range 为价格缩放参数，color 为轴文本颜色。 返回值说明：返回 SVG 片段列表。 注意事项：左轴刻度对应 A500。 """ lines = [] for tick_value in ticks: y_position = margin_top + chart_height - ((tick_value - min_close) / close_range) * chart_height lines.append('' % (margin_left, y_position, margin_left + chart_width, y_position)) lines.append('%.2f' % (margin_left - 10, y_position + 4, color, tick_value)) return lines def _render_right_axis(ticks: Sequence[float], right_axis: int, margin_top: int, chart_height: int, min_close: float, close_range: float, color: str) -> List[str]: """ 功能说明：渲染右侧纵轴刻度。 参数说明：ticks 为价格刻度，right_axis 为右轴横坐标，margin_top 和 chart_height 为绘图区参数，min_close 和 close_range 为价格缩放参数，color 为轴文本颜色。 返回值说明：返回 SVG 片段列表。 注意事项：右轴刻度对应沪深 300。 """ lines = [] for tick_value in ticks: y_position = margin_top + chart_height - ((tick_value - min_close) / close_range) * chart_height lines.append('%.2f' % (right_axis + 10, y_position + 4, color, tick_value)) return lines def _render_x_axis(ticks: Sequence[Tuple[int, date]], coordinates: Sequence[Tuple[float, float]], bottom: int) -> List[str]: """ 功能说明：渲染横轴日期刻度和文本。 参数说明：ticks 为日期刻度，coordinates 为已缩放坐标，bottom 为横轴纵坐标。 返回值说明：返回 SVG 片段列表。 注意事项：tick 索引必须存在于 coordinates 中。 """ lines = [] for index, tick_date in ticks: x_position = coordinates[index][0] lines.append('' % (x_position, bottom, x_position, bottom + 6)) lines.append('%s' % (x_position, bottom + 26, tick_date.isoformat())) return lines if __name__ == "__main__": main()