File-based Integrations

File-based 表示 vendor 以文件而不是 API 提供数据。当前文件型 integration 有两条处理路径：已迁移的 vendor 经过 ELT；未迁移的 vendor 仍由 Retail API 直接读取 legacy 文件。不要把“文件型”与“Retail API 解析文件”视为同义词。

已迁移到 ELT 的处理方式

flowchart LR
    V[Vendor SFTP / RC SFTP] --> C[Collector]
    C --> RAW[(raw S3)]
    RAW --> L[file-based-integration-lambda]
    L --> N[(normalized S3)]
    N --> Q[SQS LoadRequest]
    Q --> E[integrations/elt]
    E --> SF[(Snowflake staging + target)]
    SF --> RA[Retail API nightly sync]
    RA --> DB[(PostgreSQL)]

各组件边界如下：

组件	负责什么	不负责什么
`Collector`	从 SFTP 获取 raw 文件并写入 S3	不做业务 mapping
`file-based-integration-lambda`	解压、解析 vendor 格式、生成 normalized CSV、发送 ELT 请求	不写 PostgreSQL，不生成最终业务模型
`integrations/elt`	将 normalized 文件加载到 Snowflake，执行 SQL transform，维护 target tables	不负责 vendor SFTP，不直接写 Retail API DB
Retail API	接收 ELT completion，读取 target tables，执行 integration sync 并写 PostgreSQL	不解析已迁移 vendor 的 raw/normalized 文件

S3 数据层

ELT 路径使用两个逻辑层：

raw/：Collector 保存的供应商原始文件，作为不可变输入。
normalized/：Lambda 输出的稳定 CSV，路径通常包含 vendor、tenant（如果需要）和 as_of_date。

normalized 文件由 LoadRequest.datasets 引用。ELT 以 vendor、dataset 和 as_of_date 为运行边界，支持重跑、backfill 和按 dataset 处理。

Snowflake 数据层

ELT 在 Snowflake 中先执行 load，再执行 SQL transform：

normalized S3 → staging tables → target tables → Retail API → PostgreSQL

target tables 是 Retail API 消费的 contract。vendor-specific 文件字段只能在 Lambda 或 transform 中处理，不能让 Retail API 依赖 vendor 文件结构。

Legacy file-based 处理方式

未迁移 vendor 仍使用旧链路：

Vendor SFTP → Collector → S3 latest/文件 → Retail API parser → PostgreSQL

这类 integration 可能依赖 Rep Code、LATEST 文件和 integrations-file-based parser。排查时应先确认 vendor 是否已经被 IntegrationType::isMigratedToElt() 标记；注意部分已迁移 vendor 仍可能生成 LATEST 供兼容或其他下游使用，因此不能仅凭 LATEST 判断 Retail API 的最终输入。

当前已完成 cutover 的 vendor

截至 2026-08-20，Retail API 明确标记为 ELT 的 vendor 是：

Altruist、Apex、Betterment、Flourish、Folio Investing、LPL、My529、Pacific Life、Raymond James、RBC、Schwab、Trust America。

integrations/elt 中存在 declaration 不代表该 vendor 已经完成 Retail API cutover；状态以 isMigratedToElt() 为准。

常见数据类型

Data type	内容
Accounts	account metadata、状态和 owner 信息
Positions	security、quantity、market value
Securities	CUSIP、ticker、name 等 security reference
Tax lots	cost basis 和 acquisition date
Transactions	trade history（是否支持取决于 vendor contract）

Rep Code 与 tenant

legacy file-based integration 通常使用 vendor 侧的 Rep Code 作为 advisor identity；但 ELT vendor 可能使用 advisor、firm、RIA、subscriber code 或整份文件，不应假设所有 vendor 都按 Rep Code 分区。

在 ELT 路径中，tenant 信息随 LoadRequest 传递，并由 Retail API 的 completion handler 按 vendor-specific 规则解析。whole-file vendor 可以没有 tenant；其他 vendor 可能需要 advisor_id、rep_code、file_code、subscriber_code 或其他字段。

创建 file-based integration

Support/Admin Center 流程

Advisor 联系 Support，确认 vendor 已经向 RightCapital 提供文件。
Support 确认 vendor 的 host type、reference/tenant 信息和数据到达情况。
在 Admin Center 创建 integration mapping。
对 legacy vendor，确认 Rep Code 对应的 S3 路径和 LATEST 文件；对 ELT vendor，确认 Lambda/ELT 支持该 vendor 及 tenant contract。
通过 Link Account 选择需要导入的 accounts，并确认后续 nightly sync。

Link Account

Advisor 打开 Client Portal → Profile → Net Worth → Link Account。
选择 active integration。
legacy vendor 从 parser 读取文件；ELT vendor 从 Snowflake target tables 获取可用 accounts。
Advisor 选择 accounts，系统创建 integration_mappings 子记录并关联 Household。

Legacy parser 说明

integrations-file-based 主要用于 legacy Retail API parser，负责 CSV、fixed-width、XML 等格式的读取、字段 mapping、数据转换和校验。已迁移 vendor 的 vendor-specific normalize 和业务 transform 应放在 Lambda/ELT，不应继续扩展 legacy parser。

Legacy 故障排查

问题	检查项
Missing data	Collector run、S3 `LATEST`、文件是否包含正确 reference/Rep Code
Stale data	vendor delivery time、S3 object timestamp、`LATEST` 是否更新、sync history
Parsing error	文件编码、delimiter、header、vendor schema 是否变化
Account 未出现	integration mapping、tenant/reference、account filtering 和 parser 日志

排查入口

已迁移 vendor：检查 raw S3 → normalized S3 → SQS → ELT worker → Snowflake target → SNS → Retail API sync。
legacy vendor：检查 Collector、LATEST 文件、Retail API parser 和 integration mapping。
Collector
Nightly Sync
Integration Architecture