迭代12: 持仓内存快照（PositionSync 10s 定时同步，请求不再穿透 QMT）
- PositionSync: 内存快照 + 10s 定时全量同步 + 失败保留旧快照 + 空快照双重确认 （/health + getAsset 账户身份）+ 幽灵持仓自动清仓（连续3轮消失 + 账户身份守卫防误清） - PositionManager.getAllPositions 改读快照，空则读穿透兜底；未注入 sync 时兼容旧穿透 - 回归 test-position-sync.mjs 35 项（纯内存 mock，含 kept 分支刷新 syncedAt 防灯灰） - 文档链: R-014 + PLAN-013 + 迭代三件套 + 数据存储设计 §11（内存不落库决策与三问标准） 需求: R-014（老师拍板: 内存不落库/读穿透兜底/幽灵自动清仓/不显示同步时间）
This commit is contained in:
@@ -18,6 +18,13 @@
|
||||
* openHolding（建仓）/ addShares（加仓）/ reduceShares（减仓）/ closeHolding（清仓转历史）；
|
||||
* - 清仓/清零/删策略后数据转历史保留（closed_at），供交易记录/复盘关联（R-007 铺路）；
|
||||
* - 对外方法签名与返回结构不变，上层 api 无感知。
|
||||
*
|
||||
* 2026-09-02 持仓快照内存化（老师拍板优化）：
|
||||
* - getAllPositions 从「请求时穿透 QMT」改为「读 PositionSync 内存快照为准」，
|
||||
* PositionSync 每 10s 全量同步（失败保留上次快照）；
|
||||
* - 缓存为空（启动未预热成功）→ 读穿透兜底：当场拉一次 QMT 并回填快照；
|
||||
* - 收益：QMT 抖动不再导致持仓页面空白（读旧快照）；策略/全部持仓/未分配
|
||||
* 三个接口去掉了每次进 tab 的 QMT HTTP 调用；前端零改动。
|
||||
*/
|
||||
|
||||
import { DataStore } from '../storage/DataStore.js';
|
||||
@@ -25,22 +32,35 @@ import { DataStore } from '../storage/DataStore.js';
|
||||
export class PositionManager {
|
||||
/**
|
||||
* @param {object} opts
|
||||
* @param {import('../data-source/data-source-types.js').DataSource} opts.dataSource 数据源（QMT REST）
|
||||
* @param {import('../data-source/data-source-types.js').DataSource} opts.dataSource 数据源（QMT REST；读穿透兜底用）
|
||||
* @param {DataStore} opts.storage 数据集存储（R-006 DataStore）
|
||||
* @param {Function} [opts.getStrategySchema] 读策略自定义字段定义的回调 (strategyId) => configSchema[]
|
||||
* （R-013：由 index.js 注入，内部经 settings.getStrategies 取；缺省返回 []）
|
||||
* @param {import('./PositionSync.js').PositionSync} [opts.positionSync] 持仓内存快照同步服务
|
||||
* （2026-09-02 注入；缺省时退回旧的请求时穿透行为，保证独立可用/测试兼容）
|
||||
*/
|
||||
constructor({ dataSource, storage, getStrategySchema }) {
|
||||
constructor({ dataSource, storage, getStrategySchema, positionSync }) {
|
||||
this.dataSource = dataSource;
|
||||
this.storage = storage;
|
||||
this.positionSync = positionSync ?? null;
|
||||
this.getStrategySchema = typeof getStrategySchema === 'function'
|
||||
? getStrategySchema
|
||||
: () => [];
|
||||
}
|
||||
|
||||
/** 全量持仓（QMT 真实数据） */
|
||||
/**
|
||||
* 全量持仓（QMT 真实数据）
|
||||
* 2026-09-02 起：以 PositionSync 内存快照为准（10s 定时同步，失败保留上次快照）；
|
||||
* 快照为空 → 读穿透当场拉一次 QMT 并回填（首启兜底）；QMT 也挂 → 向上抛错（前端 LoadState 重试）。
|
||||
* 未注入 positionSync 时保持旧行为（穿透），不破坏独立构造场景。
|
||||
*/
|
||||
async getAllPositions() {
|
||||
return this.dataSource.getPositions();
|
||||
if (!this.positionSync) return this.dataSource.getPositions();
|
||||
const cached = this.positionSync.getSnapshot();
|
||||
if (cached.length > 0) return cached;
|
||||
const positions = await this.dataSource.getPositions();
|
||||
this.positionSync.backfill(positions);
|
||||
return positions;
|
||||
}
|
||||
|
||||
/** 单只持仓 */
|
||||
|
||||
@@ -0,0 +1,229 @@
|
||||
/**
|
||||
* PositionSync —— 全量持仓内存快照同步（2026-09-02 讨论定稿）
|
||||
*
|
||||
* 背景：策略持仓 / 全部持仓 / 未分配三个接口原先在请求时穿透 QMT
|
||||
* （PositionManager.getAllPositions 实时拉 /trade/positions），带来两个问题：
|
||||
* ① QMT 一抖（超时/掉线）所有持仓页面当场空白；
|
||||
* ② 每次进 tab 都打一次 QMT HTTP。
|
||||
*
|
||||
* 方案（老师拍板）：服务端建内存快照，以快照为准：
|
||||
* - 启动预热一次，之后每 10s 全量拉 QMT 持仓 → 校验 → 整体替换内存快照；
|
||||
* - 不落库（内存管理，讨论明确：不建缓存表、不复用 strategy_holdings）——
|
||||
* 持仓快照随时可用一次调用重拿全，落库只会引入「过期数据冒充实时的说谎风险」；
|
||||
* - 同步失败保留上次快照（不清空、不报错给读方）；
|
||||
* - 空快照需双重确认（/health 可用 + getAsset 账户身份可识别）才接受为「真清仓」，
|
||||
* 否则视为 QMT 异常（如未登录），保留旧快照；
|
||||
* - 缓存为空时由 PositionManager 读穿透兜底（当场拉一次 QMT 并回填 backfill）；
|
||||
*
|
||||
* 幽灵持仓自动清仓（2026-09-02 老师选定）：
|
||||
* - QMT 快照中消失的 code（本地 strategy_holdings 仍有当前持仓），连续 3 轮同步
|
||||
* （约 30s）仍消失 → 该 code 全部策略的当前持仓 closeHolding 转历史（不物理删除）；
|
||||
* - 防抖护栏：
|
||||
* a) 账户身份守卫——每轮同步顺带 getAsset 取 accountId；身份未知（接口失败/未登录）
|
||||
* 当轮跳过清仓判定（不累计计数）；accountId 变化（R-004 连接热切换/换账户）时
|
||||
* 重置防抖计数，防止 A 账户持仓被 B 账户快照误清；
|
||||
* b) 部分减持不触发（QMT 仍有该 code，只是 volume 变小）——账实差额由前端
|
||||
* 「未分配为负」暴露，属产品已知行为，不在本模块处理。
|
||||
*
|
||||
* 设计对齐：TradeSync 同构（启动预热 + 定时 + 失败容忍）；技术约束-011 测试隔离不受影响
|
||||
* （本模块无持久化，测试无需隔离数据目录）。
|
||||
*/
|
||||
|
||||
const SYNC_INTERVAL_MS = 10 * 1000; // 同步间隔（老师定：10 秒）
|
||||
const GHOST_CLOSE_ROUNDS = 3; // 幽灵防抖：连续 N 轮消失才自动清仓（约 30s）
|
||||
|
||||
export class PositionSync {
|
||||
/**
|
||||
* @param {object} opts
|
||||
* @param {object} opts.runtime { dataSource, storage } —— dataSource 拉实盘，storage 做幽灵清仓
|
||||
* @param {object} [opts.logger]
|
||||
* @param {number} [opts.intervalMs] 同步间隔（测试可调小）
|
||||
* @param {number} [opts.ghostRounds] 幽灵清仓防抖轮数（测试可调小）
|
||||
*/
|
||||
constructor({ runtime, logger, intervalMs = SYNC_INTERVAL_MS, ghostRounds = GHOST_CLOSE_ROUNDS } = {}) {
|
||||
this.runtime = runtime;
|
||||
this.logger = logger;
|
||||
this.intervalMs = intervalMs;
|
||||
this.ghostRounds = Math.max(1, ghostRounds);
|
||||
/** 内存快照：Position[]（QmtBridgeRestDataSource.mapPosition 语义化输出；只读约定） */
|
||||
this.snapshot = [];
|
||||
/** 最近一次成功同步时间（毫秒；0 = 尚未同步成功过） */
|
||||
this.syncedAt = 0;
|
||||
this.timer = null;
|
||||
this.mounted = false;
|
||||
this.syncing = false;
|
||||
/** 幽灵防抖计数：code → 连续从 QMT 快照消失的轮数 */
|
||||
this._ghostMiss = new Map();
|
||||
/** 账户身份守卫：最近一次 getAsset 的 accountId（连接热切换时重置防抖） */
|
||||
this._lastAccountId = null;
|
||||
this.stats = { syncCount: 0, failCount: 0, lastError: '', closedGhosts: 0, lastSyncedAt: 0 };
|
||||
}
|
||||
|
||||
/** 启动：立即预热一次 + 定时同步 */
|
||||
start() {
|
||||
if (this.mounted) return;
|
||||
this.mounted = true;
|
||||
this.syncNow().catch((e) => {
|
||||
this.logger?.warn?.('[one-divine-lot] PositionSync 启动预热失败: ' + (e?.message ?? e));
|
||||
});
|
||||
this.timer = setInterval(() => {
|
||||
this.syncNow().catch(() => { /* syncNow 内部已容错 */ });
|
||||
}, this.intervalMs);
|
||||
this.logger?.info?.('[one-divine-lot] PositionSync 启动（10s 内存快照同步 + 幽灵持仓自动清仓防抖 ' + this.ghostRounds + ' 轮）');
|
||||
}
|
||||
|
||||
/** 停止（插件释放时）；内存快照保留（读方在停止后仍可消费最后快照） */
|
||||
stop() {
|
||||
this.mounted = false;
|
||||
if (this.timer) {
|
||||
clearInterval(this.timer);
|
||||
this.timer = null;
|
||||
}
|
||||
}
|
||||
|
||||
/** 当前内存快照（可能为空数组 = 尚未同步成功过；只读约定，勿改写） */
|
||||
getSnapshot() {
|
||||
return this.snapshot;
|
||||
}
|
||||
|
||||
/** 最近同步时间（毫秒；0 = 从未成功） */
|
||||
getSyncedAt() {
|
||||
return this.syncedAt;
|
||||
}
|
||||
|
||||
/**
|
||||
* 读穿透回填（PositionManager 缓存未命中时调用）：把当场拉到的持仓填入内存快照。
|
||||
* 只在非空时回填（空数组无法区分真清仓与异常，交给定时同步的健康确认逻辑）。
|
||||
* @param {Array} positions 语义化持仓列表
|
||||
*/
|
||||
backfill(positions) {
|
||||
if (Array.isArray(positions) && positions.length > 0) {
|
||||
this.snapshot = positions;
|
||||
this.syncedAt = Date.now();
|
||||
this.stats.lastSyncedAt = this.syncedAt;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 同步一次：拉 QMT 全量持仓 → 校验 → 整体替换内存快照 → 幽灵清仓判定。
|
||||
* 任何失败只记统计，不动内存（读方继续消费上次快照）。
|
||||
* @returns {Promise<{positions:number, closed:Array}|{kept:true, reason:string}|null>}
|
||||
*/
|
||||
async syncNow() {
|
||||
// mounted 不拦手动同步（测试/热切换后手动触发均可用）；syncing 只防重入
|
||||
if (this.syncing) return null;
|
||||
const { dataSource } = this.runtime;
|
||||
if (!dataSource) return null;
|
||||
this.syncing = true;
|
||||
try {
|
||||
const positions = await dataSource.getPositions();
|
||||
if (!Array.isArray(positions)) throw new Error('持仓响应格式无效（非数组）');
|
||||
|
||||
if (positions.length === 0) {
|
||||
// 空快照：区分「真清仓」与「QMT 异常」。双重确认都通过才接受清空，否则保留旧快照。
|
||||
const alive = await dataSource.isAvailable().catch(() => false);
|
||||
const accountId = await this._fetchAccountId().catch(() => null);
|
||||
if (!alive || !accountId) {
|
||||
this.stats.failCount++;
|
||||
this.stats.lastError = !alive ? '空快照且 QMT health 不可用' : '空快照且账户身份不可识别（疑似未登录）';
|
||||
this.logger?.debug?.('[one-divine-lot] PositionSync 空快照不接受（' + this.stats.lastError + '），保留旧快照 ' + this.snapshot.length + ' 行');
|
||||
// 同步动作本身成功（QMT 应答了，只是不接受数据）→ 刷新 syncedAt，指示灯不因此灰
|
||||
this.syncedAt = Date.now();
|
||||
this.stats.lastSyncedAt = this.syncedAt;
|
||||
return { kept: true, reason: this.stats.lastError };
|
||||
}
|
||||
// 真清仓：接受空快照（幽灵清仓会把本地当前持仓全部转历史）
|
||||
}
|
||||
|
||||
// ① 整体替换内存快照（校验通过才动内存）
|
||||
this.snapshot = positions;
|
||||
this.syncedAt = Date.now();
|
||||
this.stats.syncCount++;
|
||||
this.stats.lastSyncedAt = this.syncedAt;
|
||||
|
||||
// ② 幽灵持仓自动清仓（带防抖；失败不影响快照）
|
||||
const closed = await this._autoCloseGhosts(positions);
|
||||
return { positions: positions.length, closed };
|
||||
} catch (e) {
|
||||
this.stats.failCount++;
|
||||
this.stats.lastError = e?.message ?? String(e);
|
||||
this.logger?.debug?.('[one-divine-lot] PositionSync 同步失败（保留旧快照 ' + this.snapshot.length + ' 行）: ' + this.stats.lastError);
|
||||
return null;
|
||||
} finally {
|
||||
this.syncing = false;
|
||||
}
|
||||
}
|
||||
|
||||
/** 取账户身份（getAsset 失败/无 accountId 返回 null） */
|
||||
async _fetchAccountId() {
|
||||
try {
|
||||
const a = await this.runtime.dataSource.getAsset();
|
||||
return a?.accountId || null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 幽灵清仓：快照中消失的 code，连续 N 轮仍消失 → 该 code 全部策略的当前持仓转历史。
|
||||
* 账户身份守卫见类注释；closeHolding 置 shares=0 + closed_at（不物理删除，历史保留）。
|
||||
* @param {Array} snapshot 本轮成功的 QMT 快照
|
||||
* @returns {Promise<Array<{code:string, strategies:string[]}>>} 本轮实际清仓的 code
|
||||
*/
|
||||
async _autoCloseGhosts(snapshot) {
|
||||
const { storage } = this.runtime;
|
||||
const closed = [];
|
||||
|
||||
// 守卫：账户身份未知 → 本轮跳过清仓判定（不累计、不清零，保守）
|
||||
const accountId = await this._fetchAccountId();
|
||||
if (!accountId) return closed;
|
||||
if (this._lastAccountId != null && accountId !== this._lastAccountId) {
|
||||
// 换账户（连接热切换）：旧计数作废，本轮直接跳过判定（连旧账户的快照都不可信）
|
||||
this._ghostMiss.clear();
|
||||
this._lastAccountId = accountId;
|
||||
this.logger?.info?.('[one-divine-lot] PositionSync 检测到账户切换（' + this._lastAccountId + ' → ' + accountId + '），本轮跳过幽灵判定');
|
||||
return closed;
|
||||
}
|
||||
this._lastAccountId = accountId;
|
||||
|
||||
let holdings;
|
||||
try {
|
||||
holdings = await storage.getCurrentHoldings(); // 全策略当前持仓（closed_at IS NULL）
|
||||
} catch (e) {
|
||||
this.logger?.debug?.('[one-divine-lot] PositionSync 读本地持仓失败，跳过幽灵判定: ' + (e?.message ?? e));
|
||||
return closed;
|
||||
}
|
||||
|
||||
const snapshotCodes = new Set(snapshot.map((p) => p.code).filter(Boolean));
|
||||
const localCodes = [...new Set(holdings.map((h) => h.code).filter(Boolean))];
|
||||
|
||||
for (const code of localCodes) {
|
||||
if (snapshotCodes.has(code)) {
|
||||
this._ghostMiss.delete(code); // QMT 仍有 → 计数复位（部分减持不在此处理）
|
||||
continue;
|
||||
}
|
||||
const miss = (this._ghostMiss.get(code) ?? 0) + 1;
|
||||
if (miss < this.ghostRounds) {
|
||||
this._ghostMiss.set(code, miss);
|
||||
continue;
|
||||
}
|
||||
// 连续 N 轮消失 → 清仓该 code 全部策略的当前持仓（转历史，不物理删除）
|
||||
try {
|
||||
const rows = holdings.filter((h) => h.code === code);
|
||||
for (const h of rows) {
|
||||
await storage.closeHolding(h.strategyId, code);
|
||||
}
|
||||
this._ghostMiss.delete(code);
|
||||
this.stats.closedGhosts += rows.length;
|
||||
closed.push({ code, strategies: rows.map((r) => r.strategyId) });
|
||||
this.logger?.warn?.(
|
||||
'[one-divine-lot] PositionSync 幽灵清仓: ' + code + ' × ' + rows.length + ' 个策略（QMT 连续 ' + this.ghostRounds + ' 轮无此持仓，自动转历史）'
|
||||
);
|
||||
} catch (e) {
|
||||
// 单码清仓失败：保留计数，下轮重试
|
||||
this.logger?.debug?.('[one-divine-lot] PositionSync 幽灵清仓失败（下轮重试）: ' + code + ' ' + (e?.message ?? e));
|
||||
}
|
||||
}
|
||||
return closed;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user