feat(FEAT-MO-005): 建立市场结构数据映射边界

2026-08-25 01:42:59 +08:00
parent a019aaa343
commit 548f9c4b8b
7 changed files with 424 additions and 5 deletions
@@ -2,6 +2,7 @@ import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';

 export type MarketOverviewStatus = 'fresh' | 'delayed' | 'partial' | 'unavailable';
 export type QuoteDirection = 'up' | 'down' | 'flat';
+export type MarketOverviewSource = 'tdx' | 'aktools' | 'mixed';

 export class MarketOverviewMetaDto {
  @ApiProperty({ type: String })
@@ -10,8 +11,8 @@ export class MarketOverviewMetaDto {
  @ApiProperty({ type: String, format: 'date-time' })
  asOf!: string;

-  @ApiProperty({ type: String, enum: ['tdx'] })
-  source!: 'tdx';
+  @ApiProperty({ type: String, enum: ['tdx', 'aktools', 'mixed'] })
+  source!: MarketOverviewSource;

  @ApiProperty({ enum: ['fresh', 'delayed', 'partial', 'unavailable'] })
  status!: MarketOverviewStatus;
@@ -0,0 +1,45 @@
+import { describe, expect, it, vi } from 'vitest';
+
+import { createAktoolsStructureClient } from './aktools-structure.client';
+
+describe('AKTools 市场结构客户端', () => {
+  it('只调用固定的行业接口并解析数组', async () => {
+    const fetchImpl = vi
+      .fn<typeof fetch>()
+      .mockResolvedValue(new Response(JSON.stringify([{ 名称: '半导体' }]), { status: 200 }));
+    const client = createAktoolsStructureClient({
+      enabled: true,
+      baseUrl: 'http://127.0.0.1:8080',
+      timeoutMs: 1000,
+      fetchImpl,
+    });
+
+    await expect(client.getSectorRows('industry')).resolves.toEqual([{ 名称: '半导体' }]);
+    expect(fetchImpl).toHaveBeenCalledWith(
+      'http://127.0.0.1:8080/api/public/stock_board_industry_spot_em',
+      expect.objectContaining({ method: 'GET' }),
+    );
+  });
+
+  it('不为未验证的 board 分类拼接任意接口', async () => {
+    const client = createAktoolsStructureClient({
+      enabled: true,
+      baseUrl: 'http://127.0.0.1:8080',
+      timeoutMs: 1000,
+      fetchImpl: vi.fn(),
+    });
+    await expect(client.getSectorRows('board')).rejects.toMatchObject({ kind: 'unavailable' });
+  });
+
+  it('服务关闭时不发起请求', async () => {
+    const fetchImpl = vi.fn<typeof fetch>();
+    const client = createAktoolsStructureClient({
+      enabled: false,
+      baseUrl: 'http://127.0.0.1:8080',
+      timeoutMs: 1000,
+      fetchImpl,
+    });
+    await expect(client.getCapitalFlowRows('industry')).rejects.toMatchObject({ kind: 'disabled' });
+    expect(fetchImpl).not.toHaveBeenCalled();
+  });
+});
@@ -0,0 +1,100 @@
+import type { AktoolsRuntimeConfig } from '../../../../../libs/config/src/runtime-config';
+
+import { AktoolsClientError } from '../aktools/aktools.client';
+import type {
+  MarketStructureSourceClient,
+  StructureCategory,
+  StructureRawRow,
+} from './market-structure.types';
+
+export interface AktoolsStructureClientOptions extends AktoolsRuntimeConfig {
+  fetchImpl?: typeof fetch;
+  maxResponseBytes?: number;
+}
+
+const DEFAULT_MAX_RESPONSE_BYTES = 2 * 1024 * 1024;
+const ENDPOINTS = Object.freeze({
+  industry: 'stock_board_industry_spot_em',
+  concept: 'stock_board_concept_spot_em',
+  capitalFlow: 'stock_sector_fund_flow_rank',
+  marketFundFlow: 'stock_market_fund_flow',
+});
+
+function isAbortError(error: unknown): boolean {
+  return error instanceof Error && error.name === 'AbortError';
+}
+
+function categoryEndpoint(category: StructureCategory): string {
+  if (category === 'industry') return ENDPOINTS.industry;
+  if (category === 'concept') return ENDPOINTS.concept;
+  throw new AktoolsClientError('unavailable', '板块分类数据源未验证');
+}
+
+function categoryFlowType(category: StructureCategory): string {
+  if (category === 'industry') return '行业资金流';
+  if (category === 'concept') return '概念资金流';
+  throw new AktoolsClientError('unavailable', '板块资金流数据源未验证');
+}
+
+export function createAktoolsStructureClient({
+  enabled,
+  baseUrl,
+  timeoutMs,
+  fetchImpl = fetch,
+  maxResponseBytes = DEFAULT_MAX_RESPONSE_BYTES,
+}: AktoolsStructureClientOptions): MarketStructureSourceClient {
+  const endpoint = `${baseUrl.replace(/\/$/, '')}/api/public/`;
+
+  async function request(
+    path: string,
+    params: Readonly<Record<string, string>> = {},
+  ): Promise<StructureRawRow[]> {
+    if (!enabled) throw new AktoolsClientError('disabled', 'AKTools 未启用');
+    const search = new URLSearchParams(params);
+    const url = `${endpoint}${path}${search.toString() ? `?${search.toString()}` : ''}`;
+    const controller = new AbortController();
+    const timeout = setTimeout(() => controller.abort(), timeoutMs);
+    try {
+      const response = await fetchImpl(url, {
+        method: 'GET',
+        headers: { accept: 'application/json' },
+        signal: controller.signal,
+      });
+      const contentLength = response.headers.get('content-length');
+      if (contentLength && Number(contentLength) > maxResponseBytes) {
+        throw new AktoolsClientError('protocol', 'AKTools 响应体超过安全上限');
+      }
+      const text = await response.text();
+      if (Buffer.byteLength(text, 'utf8') > maxResponseBytes) {
+        throw new AktoolsClientError('protocol', 'AKTools 响应体超过安全上限');
+      }
+      if (!response.ok) throw new AktoolsClientError('unavailable', 'AKTools 服务不可用');
+      let value: unknown;
+      try {
+        value = JSON.parse(text);
+      } catch {
+        throw new AktoolsClientError('protocol', 'AKTools 返回格式错误');
+      }
+      if (!Array.isArray(value) || value.some((item) => !item || typeof item !== 'object')) {
+        throw new AktoolsClientError('protocol', 'AKTools 返回数据结构错误');
+      }
+      return value as StructureRawRow[];
+    } catch (error) {
+      if (error instanceof AktoolsClientError) throw error;
+      if (isAbortError(error)) throw new AktoolsClientError('timeout', 'AKTools 请求超时');
+      throw new AktoolsClientError('network', 'AKTools 网络请求失败');
+    } finally {
+      clearTimeout(timeout);
+    }
+  }
+
+  return {
+    getSectorRows: async (category) => request(categoryEndpoint(category)),
+    getCapitalFlowRows: async (category) =>
+      request(ENDPOINTS.capitalFlow, {
+        indicator: '今日',
+        sector_type: categoryFlowType(category),
+      }),
+    getMarketFundFlowRows: () => request(ENDPOINTS.marketFundFlow),
+  };
+}
@@ -0,0 +1,84 @@
+import { describe, expect, it } from 'vitest';
+
+import { mapCapitalFlow, mapSectorRanking, mapTurnoverPoints } from './market-structure.mapper';
+
+describe('市场结构数据映射', () => {
+  it('映射行业排行并保留涨跌幅与成交额', () => {
+    expect(
+      mapSectorRanking(
+        {
+          代码: 'BK0001',
+          名称: '半导体',
+          涨跌幅: '2.5',
+          成交额: '123456789',
+          上涨家数: '20',
+          下跌家数: '4',
+          领涨股: '测试股份',
+        },
+        'industry',
+        '20260825',
+      ),
+    ).toEqual({
+      category: 'industry',
+      code: 'BK0001',
+      name: '半导体',
+      changePercent: 2.5,
+      turnoverAmount: 1.23,
+      advancingCount: 20,
+      decliningCount: 4,
+      leader: '测试股份',
+      tradeDate: '20260825',
+    });
+  });
+
+  it('拒绝未知分类和缺少代码名称的排行记录', () => {
+    expect(mapSectorRanking({ code: 'BK0001', name: '测试' }, 'board', '20260825')).toMatchObject({
+      category: 'board',
+      code: 'BK0001',
+    });
+    expect(mapSectorRanking({ 名称: '无代码' }, 'industry', '20260825')).toBeNull();
+  });
+
+  it('只使用上游明确的净流入字段，不用成交额推导资金', () => {
+    expect(
+      mapCapitalFlow(
+        {
+          名称: '半导体',
+          '今日主力净流入-净额': '100000000',
+          成交额: '500000000',
+        },
+        'industry',
+        '20260825',
+      ),
+    ).toMatchObject({
+      name: '半导体',
+      netInflow: 1,
+      netOutflow: null,
+      turnoverAmount: 5,
+      flowDefinition: '今日主力净流入-净额',
+    });
+    expect(mapCapitalFlow({ 名称: '未知', 成交额: '100' }, 'industry', '20260825')).toMatchObject({
+      netInflow: null,
+      netOutflow: null,
+      flowDefinition: 'unavailable',
+    });
+  });
+
+  it('拒绝倒序、重复和负数成交额点并计算累计值', () => {
+    expect(
+      mapTurnoverPoints([
+        { 时间: '2026-08-25 09:30:00', 成交额: '100000000' },
+        { 时间: '2026-08-25 09:35:00', 成交额: '200000000' },
+        { 时间: '2026-08-25 09:35:00', 成交额: '300000000' },
+        { 时间: '2026-08-25 09:34:00', 成交额: '400000000' },
+        { 时间: '2026-08-25 09:40:00', 成交额: '-100000000' },
+      ]),
+    ).toEqual({
+      points: [
+        { time: '2026-08-25 09:30:00', amount: 1, cumulativeAmount: 1 },
+        { time: '2026-08-25 09:35:00', amount: 2, cumulativeAmount: 3 },
+      ],
+      invalidCount: 3,
+    });
+  });
+});
@@ -0,0 +1,120 @@
+import type {
+  CapitalFlowRecord,
+  SectorRankingRecord,
+  StructureCategory,
+  StructureRawRow,
+  TurnoverMappingResult,
+  TurnoverPoint,
+} from './market-structure.types';
+
+function toNumber(value: unknown): number | null {
+  if (typeof value === 'number') return Number.isFinite(value) ? value : null;
+  if (typeof value !== 'string' || !value.trim()) return null;
+  const parsed = Number(value.replaceAll(',', ''));
+  return Number.isFinite(parsed) ? parsed : null;
+}
+
+function toNonNegativeNumber(value: unknown): number | null {
+  const parsed = toNumber(value);
+  return parsed !== null && parsed >= 0 ? parsed : null;
+}
+
+function toText(value: unknown): string | null {
+  if (typeof value !== 'string') return null;
+  const text = value.trim();
+  return text ? text : null;
+}
+
+function firstValue(row: StructureRawRow, keys: readonly string[]): unknown {
+  for (const key of keys) {
+    if (row[key] !== undefined) return row[key];
+  }
+  return undefined;
+}
+
+function amountInYi(value: unknown): number | null {
+  const parsed = toNonNegativeNumber(value);
+  return parsed === null ? null : Math.round((parsed / 100_000_000) * 100) / 100;
+}
+
+function dateValue(value: unknown): string | null {
+  const text = toText(value);
+  return text && /^\d{8}$/.test(text) ? text : text;
+}
+
+export function mapSectorRanking(
+  row: StructureRawRow,
+  category: StructureCategory,
+  tradeDate: string | null,
+): SectorRankingRecord | null {
+  const code = toText(firstValue(row, ['代码', '板块代码', 'code', 'f12']));
+  const name = toText(firstValue(row, ['名称', '板块名称', 'name', 'f14']));
+  if (!code || !name) return null;
+
+  return {
+    category,
+    code,
+    name,
+    changePercent: toNumber(firstValue(row, ['涨跌幅', '今日涨跌幅', 'changePercent', 'f3'])),
+    turnoverAmount: amountInYi(firstValue(row, ['成交额', 'turnoverAmount', 'f48'])),
+    advancingCount: toNonNegativeNumber(firstValue(row, ['上涨家数', 'advancingCount'])),
+    decliningCount: toNonNegativeNumber(firstValue(row, ['下跌家数', 'decliningCount'])),
+    leader: toText(firstValue(row, ['领涨股', '主力净流入最大股', 'leader'])),
+    tradeDate: dateValue(tradeDate),
+  };
+}
+
+export function mapCapitalFlow(
+  row: StructureRawRow,
+  category: StructureCategory,
+  tradeDate: string | null,
+): CapitalFlowRecord {
+  const definitionKeys = ['今日主力净流入-净额', '主力净流入-净额', 'netInflow'] as const;
+  const netInflowRaw = firstValue(row, definitionKeys);
+  const netInflow = toNumber(netInflowRaw);
+  const netOutflow = toNumber(firstValue(row, ['netOutflow', '净流出', '净流出额']));
+  const flowDefinition =
+    netInflowRaw !== undefined
+      ? (definitionKeys.find((key) => row[key] !== undefined) ?? 'unknown')
+      : 'unavailable';
+
+  return {
+    category,
+    code: toText(firstValue(row, ['代码', '板块代码', 'code', 'f12'])),
+    name: toText(firstValue(row, ['名称', '板块名称', 'name', 'f14'])) ?? '未知板块',
+    netInflow: netInflow === null ? null : Math.round((netInflow / 100_000_000) * 100) / 100,
+    netOutflow: netOutflow === null ? null : Math.round((netOutflow / 100_000_000) * 100) / 100,
+    turnoverAmount: amountInYi(firstValue(row, ['成交额', 'turnoverAmount', 'f48'])),
+    tradeDate: dateValue(tradeDate),
+    flowDefinition,
+  };
+}
+
+function parseTime(row: StructureRawRow): string | null {
+  return toText(firstValue(row, ['时间', '日期时间', '时间戳', 'time', 'datetime', 'Date']));
+}
+
+export function mapTurnoverPoints(rows: readonly StructureRawRow[]): TurnoverMappingResult {
+  const points: TurnoverPoint[] = [];
+  const seen = new Set<string>();
+  let invalidCount = 0;
+
+  for (const row of rows) {
+    const time = parseTime(row);
+    const amount = amountInYi(firstValue(row, ['成交额', 'amount', 'Amount']));
+    if (
+      !time ||
+      amount === null ||
+      seen.has(time) ||
+      (points.length > 0 && time <= points.at(-1)!.time)
+    ) {
+      invalidCount += 1;
+      continue;
+    }
+    seen.add(time);
+    const previous = points.at(-1)?.cumulativeAmount ?? 0;
+    points.push({ time, amount, cumulativeAmount: Math.round((previous + amount) * 100) / 100 });
+  }
+
+  return { points, invalidCount };
+}
@@ -0,0 +1,45 @@
+export type StructureCategory = 'industry' | 'concept' | 'board';
+
+export type StructureSort = 'rise' | 'fall' | 'turnover' | 'inflow' | 'outflow';
+
+export type StructureRawRow = Readonly<Record<string, unknown>>;
+
+export interface SectorRankingRecord {
+  category: StructureCategory;
+  code: string;
+  name: string;
+  changePercent: number | null;
+  turnoverAmount: number | null;
+  advancingCount: number | null;
+  decliningCount: number | null;
+  leader: string | null;
+  tradeDate: string | null;
+}
+
+export interface CapitalFlowRecord {
+  category: StructureCategory;
+  code: string | null;
+  name: string;
+  netInflow: number | null;
+  netOutflow: number | null;
+  turnoverAmount: number | null;
+  tradeDate: string | null;
+  flowDefinition: string;
+}
+
+export interface TurnoverPoint {
+  time: string;
+  amount: number;
+  cumulativeAmount: number;
+}
+
+export interface TurnoverMappingResult {
+  points: TurnoverPoint[];
+  invalidCount: number;
+}
+
+export interface MarketStructureSourceClient {
+  getSectorRows(category: StructureCategory): Promise<StructureRawRow[]>;
+  getCapitalFlowRows(category: StructureCategory): Promise<StructureRawRow[]>;
+  getMarketFundFlowRows(): Promise<StructureRawRow[]>;
+}