异常教程
RxJava 处理金融市场数据：第 1 部分
💡一则或许对你有用的小广告
欢迎加入小哈的星球 ，你将获得：专属的项目实战 / 1v1 提问 / Java 学习路线 / 学习打卡 / 每月赠书 / 社群讨论
- 新项目:《从零手撸：仿小红书（微服务架构）》 正在持续爆肝中，基于
Spring Cloud Alibaba + Spring Boot 3.x + JDK 17...，点击查看项目介绍 ;- 《从零手撸：前后端分离博客项目（全栈开发）》 2 期已完结，演示链接： http://116.62.199.48/ ;
截止目前， 星球 内专栏累计输出 82w+ 字，讲解图 3441+ 张，还在持续爆肝中.. 后续还会上新更多项目，目标是将 Java 领域典型的项目都整一波，如秒杀系统, 在线商城, IM 即时通讯，权限管理，Spring Cloud Alibaba 微服务等等，已有 2800+ 小伙伴加入学习 ，欢迎点击围观
在我最近的一个项目中，我使用 Iteractive Brokers Java API 自动化了一个交易策略， RxJava 是处理大量实时和历史数据的完美配套框架。阅读 RxJava 的文档和示例可能会令人生畏且非常抽象，这里是一个如何使用它以及它可以做什么的实际示例：
- 将市场数据提供给 RxJava marketDataObservable 类
- 使用 groupBy、平面图和缓冲区运算符将报价数据聚合到 1 分钟柱
1. 将市场数据提供给 RxJava marketDataObservable 类
public void subscribeRealTimeData(Instrument instrument) {
controller.reqTopMktData(instrument.ibContract, "232", false, new ApiController.ITopMktDataHandler()
{
@Override
public void tickPrice(TickType tickType, double price, int canAutoExecute) {
if (tickType == TickType.ASK)
{
log.info("IB tick " + new Date() + " price " + price);
LivePriceEvent priceEvent = new LivePriceEvent(System.currentTimeMillis(), instrument, new BigDecimal(price).setScale(3, RoundingMode.UP));
marketDataObservable.push(priceEvent);
}
}
现在每次报价从 IB 到达时，它都会被推送到我们的 Obseravble。现在我们可以根据需要使用 RxJava Observable 的不同运算符折叠数据
2. 将报价数据汇总到 1 分钟柱
public void subscribeRealTimeData(Instrument instrument) {
controller.reqTopMktData(instrument.ibContract, "232", false, new ApiController.ITopMktDataHandler()
{
@Override
public void tickPrice(TickType tickType, double price, int canAutoExecute) {
if (tickType == TickType.ASK)
{
log.info("IB tick " + new Date() + " price " + price);
LivePriceEvent priceEvent = new LivePriceEvent(System.currentTimeMillis(), instrument, new BigDecimal(price).setScale(3, RoundingMode.UP));
marketDataObservable.push(priceEvent);
}
}
3. 针对 IB 演示提要运行分钟柱聚合器
只需按照 github 上的说明进行操作
public void subscribeRealTimeData(Instrument instrument) {
controller.reqTopMktData(instrument.ibContract, "232", false, new ApiController.ITopMktDataHandler()
{
@Override
public void tickPrice(TickType tickType, double price, int canAutoExecute) {
if (tickType == TickType.ASK)
{
log.info("IB tick " + new Date() + " price " + price);
LivePriceEvent priceEvent = new LivePriceEvent(System.currentTimeMillis(), instrument, new BigDecimal(price).setScale(3, RoundingMode.UP));
marketDataObservable.push(priceEvent);
}
}
你应该看到这样的东西
public void subscribeRealTimeData(Instrument instrument) {
controller.reqTopMktData(instrument.ibContract, "232", false, new ApiController.ITopMktDataHandler()
{
@Override
public void tickPrice(TickType tickType, double price, int canAutoExecute) {
if (tickType == TickType.ASK)
{
log.info("IB tick " + new Date() + " price " + price);
LivePriceEvent priceEvent = new LivePriceEvent(System.currentTimeMillis(), instrument, new BigDecimal(price).setScale(3, RoundingMode.UP));
marketDataObservable.push(priceEvent);
}
}