feat(FEAT-MO-005): 增加行业题材资金流向接口

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

@@ -1,14 +1,15 @@
 import { Inject, Injectable } from '@nestjs/common';
 
 import type { MarketOverviewStatus } from '../market-overview.dto';
-import { mapSectorRanking } from './market-structure.mapper';
+import { mapCapitalFlow, mapSectorRanking } from './market-structure.mapper';
 import type {
+  CapitalFlowRecord,
   SectorRankingRecord,
   StructureCategory,
   StructureSort,
   MarketStructureSourceClient,
 } from './market-structure.types';
-import type { SectorRankingsResponseDto } from './market-structure.dto';
+import type { CapitalFlowResponseDto, SectorRankingsResponseDto } from './market-structure.dto';
 
 export const MARKET_STRUCTURE_SOURCE_PORT = Symbol('MARKET_STRUCTURE_SOURCE_PORT');
 
@@ -18,8 +19,15 @@ export interface SectorRankingQuery {
   limit: number;
 }
 
+export interface CapitalFlowQuery {
+  category: StructureCategory;
+  sort: Extract<StructureSort, 'inflow' | 'outflow'>;
+  limit: number;
+}
+
 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',
@@ -40,6 +48,25 @@ export function parseSectorRankingQuery(
   };
 }
 
+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 snapshotMeta(
   source: 'aktools',
 ): Pick<SectorRankingsResponseDto, 'snapshotId' | 'asOf' | 'source'> {
@@ -72,6 +99,32 @@ function statusFor(items: readonly SectorRankingRecord[], rawCount: number): Mar
   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';
+}
+
 @Injectable()
 export class MarketStructureService {
   constructor(
@@ -114,6 +167,38 @@ export class MarketStructureService {
     }
   }
 
+  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: ['资金流向数据不可用'],
+      };
+    }
+  }
+
   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;