359 lines
19 KiB
Python
359 lines
19 KiB
Python
"""策略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_chart/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 """<svg xmlns="http://www.w3.org/2000/svg" width="{width}" height="{height}" viewBox="0 0 {width} {height}">
|
||
<rect width="100%" height="100%" fill="#ffffff" />
|
||
<text x="{center_x}" y="32" text-anchor="middle" font-size="22" font-weight="700" fill="#111827">{title}</text>
|
||
<text x="{center_x}" y="56" text-anchor="middle" font-size="13" fill="#6b7280">{subtitle}</text>
|
||
<circle cx="{legend_left_x}" cy="74" r="5" fill="{left_color}" />
|
||
<text x="{legend_left_text_x}" y="78" font-size="13" fill="#374151">{left_name}（左轴）</text>
|
||
<circle cx="{legend_right_x}" cy="74" r="5" fill="{right_color}" />
|
||
<text x="{legend_right_text_x}" y="78" font-size="13" fill="#374151">{right_name}（右轴）</text>
|
||
{y_axis}
|
||
<line x1="{left_axis}" y1="{top}" x2="{left_axis}" y2="{bottom}" stroke="#374151" />
|
||
<line x1="{right_axis}" y1="{top}" x2="{right_axis}" y2="{bottom}" stroke="#374151" />
|
||
<line x1="{left_axis}" y1="{bottom}" x2="{right_axis}" y2="{bottom}" stroke="#374151" />
|
||
{x_axis}
|
||
<polyline fill="none" stroke="{left_color}" stroke-width="2.4" points="{left_polyline}" />
|
||
<polyline fill="none" stroke="{right_color}" stroke-width="2.4" points="{right_polyline}" />
|
||
<text x="{center_x}" y="{footer_y}" text-anchor="middle" font-size="12" fill="#9ca3af">数据来源：sidecar / 腾讯财经；双轴按首日价格归一化，等距离代表等涨跌比例</text>
|
||
</svg>
|
||
""".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_chart/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('<line x1="%d" y1="%.2f" x2="%d" y2="%.2f" stroke="#e5e7eb" />' % (margin_left, y_position, margin_left + chart_width, y_position))
|
||
lines.append('<text x="%d" y="%.2f" text-anchor="end" font-size="12" fill="%s">%.2f (%+.1f%%)</text>' % (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('<text x="%d" y="%.2f" text-anchor="start" font-size="12" fill="%s">%.2f (%+.1f%%)</text>' % (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('<line x1="%d" y1="%.2f" x2="%d" y2="%.2f" stroke="#e5e7eb" />' % (margin_left, y_position, margin_left + chart_width, y_position))
|
||
lines.append('<text x="%d" y="%.2f" text-anchor="end" font-size="12" fill="%s">%.2f</text>' % (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('<text x="%d" y="%.2f" text-anchor="start" font-size="12" fill="%s">%.2f</text>' % (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('<line x1="%.2f" y1="%d" x2="%.2f" y2="%d" stroke="#d1d5db" />' % (x_position, bottom, x_position, bottom + 6))
|
||
lines.append('<text x="%.2f" y="%d" text-anchor="middle" font-size="12" fill="#4b5563">%s</text>' % (x_position, bottom + 26, tick_date.isoformat()))
|
||
return lines
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |