ppai/polymarket/libs/syncData.js

import 'dotenv/config';

import path from "node:path";
import { fileURLToPath } from "node:url";

import Logs from "./logs.js";
import { setData } from "./cache.js";
import getDateInTimezone from "./getDateInTimezone.js";

import { getSoccerSports, getEvents, platformPost, platformGet, MarketWsClient } from "./polymarketClient.js";
import { parseMarkets, parseOddsAsk, parseOddsBid, parseIorDetail } from "./parseMarkets.js";

const __filename = fileURLToPath(import.meta.url);
const __dirname = path.dirname(__filename);

const IS_DEV = process.env.NODE_ENV == 'development';
const IS_BID = process.env.PPAI_RUN_MODE == 'BID';

const eventsCacheFile = path.join(__dirname, "../cache/polymarketEventsCache.json");
const marketsDataFile = path.join(__dirname, "../cache/polymarketMarketsCache.json");

const GLOBAL_DATA = {
  eventsData: {},
  marketsData: {},
  clobTokenMap: {},
  filteredLeagues: [],
  relatedGames: [],
  marketWsClient: null,
  wsClientData: null,
  lastChangeTime: 0,
};

// /**
//  * 获取有效联赛
//  * @param {*} marketsList
//  * @param {*} soccerSports
//  * @returns {Array}
//  */
// const getLeagues = (marketsList, soccerSports) => {
//   const soccerSportsMap = new Map(soccerSports.map(item => [+item.series, item]));
//   const leaguesList = marketsList.map(item => {
//     const { leagueId: id, leagueName: name } = item;
//     const sport = soccerSportsMap.get(+id)?.sport;
//     return { id, name, sport };
//   });
//   // 去重并排序
//   const leaguesMap = new Map(leaguesList.map(item => [item.id, item]));
//   return Array.from(leaguesMap.values()).sort((a, b) => a.id - b.id);
// }

// /**
//  * 获取足球联赛
//  * @returns {Promise}
//  */
// const getSoccerSports = async () => {
//   return fetchMarketData({ url: "/sports" })
//   .then(sportsData => {
//     return sportsData.filter(item => {
//       const { tags } = item;
//       const tagIds = tags.split(",").map(item => +item);
//       return tagIds.includes(100350);
//     });
//   });
// }

// /**
//  * 提交联赛和比赛数据
//  * @param {string} platform - 平台名称
//  * @param {string} url - 请求地址
//  * @param {Object} data - 请求数据
//  * @returns {Promise}
//  */
// const submitLeaguesAndGames = async ({ leagues, games }) => {
//   const submitLeagues = platformPost('/api/platforms/update_leagues', { platform: 'polymarket', leagues });
//   const submitGames = platformPost('/api/platforms/update_games', { platform: 'polymarket', games });
//   return Promise.all([submitLeagues, submitGames]);
// }

// /**
//  * 更新联赛和比赛数据
//  */
// const updateLeaguesAndGames = (marketsData) => {
//   const marketsList = Object.values(marketsData);
//   getSoccerSports()
//   .then(soccerSports => {
//     const leagues = getLeagues(marketsList, soccerSports);
//     const games = marketsList.map(game => {
//       const { marketsData, ...rest } = game;
//       return rest;
//     }).sort((a, b) => a.timestamp - b.timestamp);
//     return { leagues, games };
//   })
//   .then(data => {
//     Logs.outDev('update leagues and games list', data.leagues.length, data.games.length);
//     return submitLeaguesAndGames(data);
//   })
//   .then(() => {
//     Logs.outDev('leagues and games list updated');
//   })
//   .catch(error => {
//     Logs.out('failed to update leagues and games list', error.message);
//   });
// }

// /**
//  * 获取过滤后的比赛数据
//  * @returns
//  */
// const updateRelatedGames = () => {
//   platformGet('/api/platforms/get_related_games', { platform: 'polymarket' })
//   .then(res => {
//     const { data: relatedGames } = res;
//     Logs.outDev('relatedGames updated', relatedGames.length);
//     GLOBAL_DATA.relatedGames = relatedGames.map(item => item.id);
//   })
//   .catch(error => {
//     Logs.out('failed to update related games', error.message);
//   })
//   .finally(() => {
//     setTimeout(() => {
//       updateRelatedGames();
//     }, 1000 * 10);
//   });
// }

/**
 * 获取赛事数据
 * @returns {Promise}
 */
const getMarketsData = async () => {
  const endDateMinStamps = Date.now() - 1000 * 60 * 60 * 2;
  const endDateMin = new Date(endDateMinStamps).toISOString();
  const tomorrowDateMinus4 = getDateInTimezone(-4, Date.now()+24*60*60*1000);
  const tomorrowGmtMinus4EndTime = new Date(`${tomorrowDateMinus4} 23:59:59 GMT-4`).getTime();
  const endDateMax = new Date(tomorrowGmtMinus4EndTime).toISOString();
  return getEvents({ endDateMin, endDateMax })
  .then(events => {
    const { eventsData } = GLOBAL_DATA;
    events.forEach(event => {
      const { id } = event;
      if (!eventsData[id]) {
        eventsData[id] = event;
      }
    });
    const eventsMap = new Map(events.map(event => [event.id, event]));
    Object.keys(eventsData).forEach(id => {
      if (!eventsMap.has(id)) {
        delete eventsData[id];
      }
    });
    setData(eventsCacheFile, eventsData);
    return parseMarkets(eventsData);
  })
  .then(marketsData => {

    // updateLeaguesAndGames(marketsData);

    const { marketsData: oldMarketsData } = GLOBAL_DATA;
    const newMarketsData = marketsData;

    const marketsDataUpdate = {
      add: [],
      remove: []
    }

    Object.keys(oldMarketsData).forEach(id => {
      if (!newMarketsData[id]) {
        delete oldMarketsData[id];
        marketsDataUpdate.remove.push(id);
      }
    });

    Object.keys(newMarketsData).forEach(id => {
      if (!oldMarketsData[id]) {
        oldMarketsData[id] = newMarketsData[id];
        marketsDataUpdate.add.push(id);
      }
    });

    const clobToken = Object.values(oldMarketsData).map(item => item.marketsData)
    .flatMap(item => Object.values(item))
    .flatMap(item => Object.values(item))
    .flatMap(item => Object.values(item.outcomes));

    const { clobTokenMap: oldClobTokenMap } = GLOBAL_DATA;
    const newClobTokenMap = new Map(clobToken.map(item => [item.id, item]));

    const clobTokenUpdate = {
      add: [],
      remove: []
    }

    Object.keys(oldClobTokenMap).forEach(id => {
      if (!newClobTokenMap.has(id)) {
        delete oldClobTokenMap[id];
        clobTokenUpdate.remove.push(id);
      }
    });

    newClobTokenMap.forEach(item => {
      const { id } = item;
      if (!oldClobTokenMap[id]) {
        oldClobTokenMap[id] = newClobTokenMap.get(id);
        clobTokenUpdate.add.push(id);
      }
    });

    return clobTokenUpdate;
  });
}

/**
 * 循环更新
 */
const updateGamesMarkets = () => {
  Logs.outDev('updateGamesMarkets');
  getMarketsData()
  .then(clobTokenUpdate => {
    const { marketWsClient } = GLOBAL_DATA;
    const { add, remove } = clobTokenUpdate;
    if (add.length > 0) {
      Logs.outDev('subscribeToTokensIds', add);
      marketWsClient?.subscribeToTokensIds(add);
    }
    if (remove.length > 0) {
      Logs.outDev('unsubscribeToTokensIds', remove);
      marketWsClient?.unsubscribeToTokensIds(remove);
    }
  })
  .catch(error => {
    Logs.out('failed to update games markets', error.message);
  })
  .finally(() => {
    setTimeout(() => {
      updateGamesMarkets();
    }, 1000 * 60);
  });
}

const syncMarketsData = () => {
  const { marketsData } = GLOBAL_DATA;
  setTimeout(() => {
    syncMarketsData();
  }, 1000 * 5);
  if (IS_DEV) {
    setData(marketsDataFile, marketsData);
  }
}

/**
 * 获取比赛盘口赔率数据
 * @returns {Object}
 */
const getGamesEvents = () => {
  const { marketsData, lastChangeTime } = GLOBAL_DATA;
  // Logs.outDev('getGamesEvents', marketsData, lastChangeTime);
  const games = Object.values(marketsData).map(item => {
    const { marketsData, ...rest } = item;
    const odds = IS_BID ? parseOddsBid(marketsData) : parseOddsAsk(marketsData);
    // const odds = parseOddsBid(marketsData);
    return { ...rest, odds };
  }).sort((a, b) => a.timestamp - b.timestamp);
  return { games, timestamp: lastChangeTime };
}

/**
 * 更新赔率数据
 */
const updateOdds = async () => {
  const { games, timestamp } = getGamesEvents();
  const expireTime = Date.now() - 30_000;
  if (!games.length || timestamp < expireTime ) {
    return Promise.resolve();
  }
  Logs.outDev('updateOdds', games, timestamp);
  return platformPost('/api/platforms/update_odds', { platform: 'polymarket', games, timestamp });
}

/**
 * 定时更新赔率数据
 */
const updateOddsLoop = () => {
  updateOdds()
  .catch(error => {
    Logs.err('failed to update odds', error.message);
  })
  .finally(() => {
    setTimeout(() => {
      updateOddsLoop();
    }, 1000 * 2);
  });
}

/**
 * 处理价格变化
 * @param {*} priceChanges
 */
const marketPriceChange = (priceChanges) => {
  let changed = false;
  priceChanges.forEach(priceChange => {
    const { asset_id, best_ask, best_bid } = priceChange;
    const { clobTokenMap } = GLOBAL_DATA;
    const clobToken = clobTokenMap[asset_id];
    if (!clobToken) {
      return;
    }
    Object.assign(clobToken, { best_ask, best_bid });
    changed = true;
  });
  if (changed) {
    GLOBAL_DATA.lastChangeTime = Date.now();
  }
}

/**
 * 处理市场数据
 */
const marketBook = (data) => {
  const { asks, bids, asset_id } = data;
  const { clobTokenMap } = GLOBAL_DATA;
  const clobToken = clobTokenMap[asset_id];
  if (!clobToken) {
    return;
  }
  const bestAskItem = asks?.reduce((minItem, current) => {
    return Number(current.price) < Number(minItem.price) ? current : minItem;
  }, asks?.[0] ?? { price: '1' });
  const best_ask = bestAskItem?.price ?? '0.001';
  const bestBidItem = bids?.reduce((maxItem, current) => {
    return Number(current.price) > Number(maxItem.price) ? current : maxItem;
  }, bids?.[0] ?? { price: '0' });
  const best_bid = bestBidItem?.price ?? '0';
  Object.assign(clobToken, { best_ask, best_bid });
  GLOBAL_DATA.lastChangeTime = Date.now();
}

/**
 * 启动同步市场数据
 */
export const startSyncMarketsData = () => {
  const marketWsClient = new MarketWsClient();
  GLOBAL_DATA.marketWsClient = marketWsClient;
  marketWsClient.connect();
  marketWsClient.on('open', () => {
    // updateRelatedGames();
    updateGamesMarkets();
    syncMarketsData();
    updateOddsLoop();
  });
  marketWsClient.on('message', data => {
    // Logs.outDev('message data:', data);
    switch(data.event_type) {
      case 'price_change':
        marketPriceChange(data.price_changes);
        break;
      case 'book':
        marketBook(data);
        break
    }
  });
  marketWsClient.on('error', error => {
    Logs.err('error', error);
  });
  marketWsClient.on('close', event => {
    Logs.outDev('close', event);
  });
}

/**
 * 获取盘口详情
 * @param {*} ior
 * @param {*} id
 * @returns
 */
export const getIorInfo = (ior, id) => {
  if (!id || !ior) {
    return Promise.reject({ cause: 400, message: 'id and ior are required', data: { id, ior } });
  }
  const { marketsData } = GLOBAL_DATA;
  const iorInfo = parseIorDetail(ior, id, marketsData);
  if (iorInfo.cause === 400) {
    return Promise.reject({ cause: 400, message: iorInfo.message, data: { id, ior } });
  }
  Logs.outDev('getIorInfo', { id, ior }, iorInfo);
  return Promise.resolve(iorInfo);
}