a-share-analysis/server/apps/api/src/market-overview/structure/market-structure.service.ts

import { Inject, Injectable, Optional } from '@nestjs/common';

import type { MarketOverviewStatus } from '../market-overview.dto';
import { TDX_CLIENT_PORT } from '../market-overview.service';
import type { TdxClient } from '../tdx/tdx.client';
import { TURNOVER_INDEXES, type TdxKlineSeries, type TdxMarketDataResult } from '../tdx/tdx.types';
import { mapCapitalFlow, mapSectorRanking } from './market-structure.mapper';
import { mapTurnoverPoints } from './market-structure.mapper';
import type {
  CapitalFlowRecord,
  SectorRankingRecord,
  StructureCategory,
  StructureSort,
  MarketStructureSourceClient,
} from './market-structure.types';
import type {
  CapitalFlowResponseDto,
  SectorRankingsResponseDto,
  TurnoverTrendResponseDto,
} from './market-structure.dto';

export const MARKET_STRUCTURE_SOURCE_PORT = Symbol('MARKET_STRUCTURE_SOURCE_PORT');

export interface SectorRankingQuery {
  category: StructureCategory;
  sort: Extract<StructureSort, 'rise' | 'fall' | 'turnover'>;
  limit: number;
}

export interface CapitalFlowQuery {
  category: StructureCategory;
  sort: Extract<StructureSort, 'inflow' | 'outflow'>;
  limit: number;
}

export interface TurnoverTrendQuery {
  interval: '5' | '15';
  tradeDate: string;
}

const CATEGORIES = new Set<StructureCategory>(['industry', 'concept', 'board']);
const SORTS = new Set<SectorRankingQuery['sort']>(['rise', 'fall', 'turnover']);
const FLOW_SORTS = new Set<CapitalFlowQuery['sort']>(['inflow', 'outflow']);

export function parseSectorRankingQuery(
  category = 'industry',
  sort = 'rise',
  limit = '10',
): SectorRankingQuery {
  if (!CATEGORIES.has(category as StructureCategory)) throw new Error('分类参数无效');
  if (!SORTS.has(sort as SectorRankingQuery['sort'])) throw new Error('排序参数无效');
  if (!/^\d+$/.test(limit)) throw new Error('数量参数无效');
  const parsedLimit = Number(limit);
  if (!Number.isSafeInteger(parsedLimit) || parsedLimit < 1 || parsedLimit > 100) {
    throw new Error('数量参数无效');
  }
  return {
    category: category as StructureCategory,
    sort: sort as SectorRankingQuery['sort'],
    limit: parsedLimit,
  };
}

export function parseCapitalFlowQuery(
  category = 'industry',
  sort = 'inflow',
  limit = '10',
): CapitalFlowQuery {
  if (!CATEGORIES.has(category as StructureCategory)) throw new Error('分类参数无效');
  if (!FLOW_SORTS.has(sort as CapitalFlowQuery['sort'])) throw new Error('排序参数无效');
  if (!/^\d+$/.test(limit)) throw new Error('数量参数无效');
  const parsedLimit = Number(limit);
  if (!Number.isSafeInteger(parsedLimit) || parsedLimit < 1 || parsedLimit > 100) {
    throw new Error('数量参数无效');
  }
  return {
    category: category as StructureCategory,
    sort: sort as CapitalFlowQuery['sort'],
    limit: parsedLimit,
  };
}

function isValidDate(value: string): boolean {
  if (!/^\d{8}$/.test(value)) return false;
  const date = new Date(`${value.slice(0, 4)}-${value.slice(4, 6)}-${value.slice(6, 8)}T00:00:00Z`);
  return (
    date.getUTCFullYear() === Number(value.slice(0, 4)) &&
    date.getUTCMonth() + 1 === Number(value.slice(4, 6)) &&
    date.getUTCDate() === Number(value.slice(6, 8))
  );
}

export function parseTurnoverTrendQuery(
  interval = '5',
  tradeDate = currentShanghaiDate(),
): TurnoverTrendQuery {
  if (interval !== '5' && interval !== '15') throw new Error('粒度参数无效');
  if (!isValidDate(tradeDate)) throw new Error('交易日期参数无效');
  return { interval, tradeDate };
}

function currentShanghaiDate(now = new Date()): string {
  const values = new Intl.DateTimeFormat('en-CA', {
    timeZone: 'Asia/Shanghai',
    year: 'numeric',
    month: '2-digit',
    day: '2-digit',
  }).formatToParts(now);
  const parts = Object.fromEntries(values.map((item) => [item.type, item.value]));
  return `${parts.year}${parts.month}${parts.day}`;
}

function snapshotMeta(
  source: 'aktools',
): Pick<SectorRankingsResponseDto, 'snapshotId' | 'asOf' | 'source'> {
  const asOf = new Date().toISOString();
  return {
    snapshotId: asOf.replace(/[-:TZ.]/g, '').slice(0, 15),
    asOf,
    source,
  };
}

function sortItems(
  items: readonly SectorRankingRecord[],
  sort: SectorRankingQuery['sort'],
): SectorRankingRecord[] {
  const valueFor = (item: SectorRankingRecord): number => {
    if (sort === 'turnover') return item.turnoverAmount ?? Number.NEGATIVE_INFINITY;
    return item.changePercent ?? Number.NEGATIVE_INFINITY;
  };
  const direction = sort === 'fall' ? 1 : -1;
  return [...items].sort((left, right) => {
    const difference = (valueFor(left) - valueFor(right)) * direction;
    return difference || left.name.localeCompare(right.name, 'zh-CN');
  });
}

function statusFor(items: readonly SectorRankingRecord[], rawCount: number): MarketOverviewStatus {
  if (!items.length) return 'unavailable';
  if (items.length < rawCount || items.some((item) => item.tradeDate === null)) return 'partial';
  return 'fresh';
}

function sortFlows(
  items: readonly CapitalFlowRecord[],
  sort: CapitalFlowQuery['sort'],
): CapitalFlowRecord[] {
  return [...items].sort((left, right) => {
    const leftValue = left.netInflow ?? Number.NEGATIVE_INFINITY;
    const rightValue = right.netInflow ?? Number.NEGATIVE_INFINITY;
    const difference = sort === 'inflow' ? rightValue - leftValue : leftValue - rightValue;
    return difference || left.name.localeCompare(right.name, 'zh-CN');
  });
}

function flowStatusFor(
  items: readonly CapitalFlowRecord[],
  rawCount: number,
): MarketOverviewStatus {
  if (!items.length) return 'unavailable';
  if (
    items.length < rawCount ||
    items.some((item) => item.flowDefinition === 'unavailable' || item.tradeDate === null)
  ) {
    return 'partial';
  }
  return 'fresh';
}

function aggregateTdxSeries(
  result: TdxMarketDataResult,
  tradeDate: string,
): { rows: Array<{ 时间: string; 成交额: number }>; marketCount: number } {
  const amountsByTime = new Map<string, number>();
  let marketCount = 0;
  for (const definition of TURNOVER_INDEXES) {
    const series: TdxKlineSeries | undefined = result.Value?.[definition.code];
    if (!series || !Array.isArray(series.Date) || !Array.isArray(series.Amount)) continue;
    if (series.Date.length === 0 || series.Date.length !== series.Amount.length) continue;
    const validRows = series.Date.flatMap((time, index) => {
      if (
        typeof time !== 'string' ||
        !time.startsWith(
          `${tradeDate.slice(0, 4)}-${tradeDate.slice(4, 6)}-${tradeDate.slice(6, 8)}`,
        )
      )
        return [];
      const amount = Number(series.Amount[index]);
      return Number.isFinite(amount) && amount >= 0 ? [{ time, amount: amount * 10_000 }] : [];
    });
    if (!validRows.length) continue;
    marketCount += 1;
    for (const row of validRows)
      amountsByTime.set(row.time, (amountsByTime.get(row.time) ?? 0) + row.amount);
  }
  return {
    rows: [...amountsByTime.entries()]
      .sort(([left], [right]) => left.localeCompare(right))
      .map(([时间, 成交额]) => ({ 时间, 成交额 })),
    marketCount,
  };
}

@Injectable()
export class MarketStructureService {
  constructor(
    @Inject(MARKET_STRUCTURE_SOURCE_PORT)
    private readonly source: MarketStructureSourceClient,
    @Optional()
    @Inject(TDX_CLIENT_PORT)
    private readonly tdxClient: TdxClient = {
      request: async <TResponse>() => ({}) as TResponse,
    },
  ) {}

  async getSectorRankings(query: SectorRankingQuery): Promise<SectorRankingsResponseDto> {
    const meta = snapshotMeta('aktools');
    try {
      const rows = await this.source.getSectorRows(query.category);
      const items = rows.flatMap((row) => {
        const mapped = mapSectorRanking(row, query.category, this.readTradeDate(row));
        return mapped ? [mapped] : [];
      });
      const sortedAll = sortItems(items, query.sort);
      const sorted = sortedAll.slice(0, query.limit);
      const status = statusFor(sortedAll, rows.length);
      return {
        ...meta,
        status,
        data:
          status === 'unavailable'
            ? null
            : { items: sorted, sort: query.sort, category: query.category },
        warnings:
          status === 'fresh'
            ? []
            : status === 'partial'
              ? ['部分行业或题材排行字段不可用']
              : ['行业排行数据不可用'],
      };
    } catch {
      return {
        ...meta,
        status: 'unavailable',
        data: null,
        warnings: ['行业排行数据不可用'],
      };
    }
  }

  async getCapitalFlow(query: CapitalFlowQuery): Promise<CapitalFlowResponseDto> {
    const meta = snapshotMeta('aktools');
    try {
      const rows = await this.source.getCapitalFlowRows(query.category);
      const items = rows.map((row) => mapCapitalFlow(row, query.category, this.readTradeDate(row)));
      const sortedAll = sortFlows(items, query.sort);
      const sorted = sortedAll.slice(0, query.limit);
      const status = flowStatusFor(sortedAll, rows.length);
      return {
        ...meta,
        status,
        data:
          status === 'unavailable'
            ? null
            : { items: sorted, sort: query.sort, category: query.category },
        warnings:
          status === 'fresh'
            ? []
            : status === 'partial'
              ? ['部分资金流向字段或交易日期不可用']
              : ['资金流向数据不可用'],
      };
    } catch {
      return {
        ...meta,
        status: 'unavailable',
        data: null,
        warnings: ['资金流向数据不可用'],
      };
    }
  }

  async getTurnoverTrend(query: TurnoverTrendQuery): Promise<TurnoverTrendResponseDto> {
    const asOf = new Date().toISOString();
    const meta = {
      snapshotId: asOf.replace(/[-:TZ.]/g, '').slice(0, 15),
      asOf,
      source: 'tdx' as const,
    };
    try {
      const result = await this.tdxClient.request<TdxMarketDataResult>('get_market_data', {
        stock_list: TURNOVER_INDEXES.map((item) => item.code),
        count: 240,
        dividend_type: 'none',
        period: `${query.interval}m`,
      });
      const aggregated = aggregateTdxSeries(result, query.tradeDate);
      const mapped = mapTurnoverPoints(aggregated.rows);
      if (!mapped.points.length) {
        return { ...meta, status: 'unavailable', data: null, warnings: ['成交额趋势数据不可用'] };
      }
      const dateStatus: MarketOverviewStatus =
        query.tradeDate === currentShanghaiDate() ? 'fresh' : 'delayed';
      const status: MarketOverviewStatus =
        aggregated.marketCount < TURNOVER_INDEXES.length || mapped.invalidCount > 0
          ? 'partial'
          : dateStatus;
      return {
        ...meta,
        status,
        data: {
          points: mapped.points,
          granularity: query.interval,
          unit: '亿元',
          marketCount: aggregated.marketCount,
          invalidPointCount: mapped.invalidCount,
        },
        warnings:
          status === 'fresh'
            ? []
            : status === 'partial'
              ? ['部分市场或成交额点不可用']
              : [`成交额交易日期为 ${query.tradeDate}，非当前日期`],
      };
    } catch {
      return { ...meta, status: 'unavailable', data: null, warnings: ['成交额趋势数据不可用'] };
    }
  }

  private readTradeDate(row: Readonly<Record<string, unknown>>): string | null {
    const value = row['日期'] ?? row['交易日期'] ?? row['tradeDate'];
    return typeof value === 'string' && /^\d{8}$/.test(value) ? value : null;
  }
}