From e50d3a160bc6cfecfcc5134cdd8e47764a95608b Mon Sep 17 00:00:00 2001 From: Tommy Date: Tue, 11 Aug 2026 12:07:44 -0400 Subject: [PATCH] Initial commit: IB auto-trading bot with multi-strategy support --- .env.example | 10 + .gitignore | 13 ++ AGENTS.md | 96 ++++++++++ README.md | 141 ++++++++++++++ bars.py | 127 +++++++++++++ close_legacy_positions.py | 120 ++++++++++++ close_legacy_positions.sh | 33 ++++ config.py | 145 +++++++++++++++ connection.py | 56 ++++++ daily_report.py | 151 +++++++++++++++ main.py | 220 ++++++++++++++++++++++ orders.py | 119 ++++++++++++ report.sh | 5 + requirements.txt | 4 + restart_bot.sh | 28 +++ start_gateway.sh | 16 ++ state.py | 320 +++++++++++++++++++++++++++++++ stop_bot_eod.sh | 16 ++ strategies/__init__.py | 7 + strategies/base.py | 351 +++++++++++++++++++++++++++++++++++ strategies/forex.py | 120 ++++++++++++ strategies/ma_cross.py | 212 +++++++++++++++++++++ strategies/mean_reversion.py | 237 +++++++++++++++++++++++ strategies/short_term.py | 216 +++++++++++++++++++++ 24 files changed, 2763 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 AGENTS.md create mode 100644 README.md create mode 100644 bars.py create mode 100644 close_legacy_positions.py create mode 100755 close_legacy_positions.sh create mode 100644 config.py create mode 100644 connection.py create mode 100644 daily_report.py create mode 100644 main.py create mode 100644 orders.py create mode 100755 report.sh create mode 100644 requirements.txt create mode 100755 restart_bot.sh create mode 100755 start_gateway.sh create mode 100644 state.py create mode 100755 stop_bot_eod.sh create mode 100644 strategies/__init__.py create mode 100644 strategies/base.py create mode 100644 strategies/forex.py create mode 100644 strategies/ma_cross.py create mode 100644 strategies/mean_reversion.py create mode 100644 strategies/short_term.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..ce94dd7 --- /dev/null +++ b/.env.example @@ -0,0 +1,10 @@ +IB_HOST=127.0.0.1 +# IB Gateway: 4002=paper, 4001=live +# TWS: 7497=paper, 7496=live +# 实盘使用 4001 时，请确认你已充分测试过策略 +IB_PORT=4001 +IB_CLIENT_ID=1 +# 你的 IB 账户号（如 U4845070） +IB_ACCOUNT= +# 仅用于启动时的日志警示，真实连接模式由端口/Gateway 登录类型决定 +IB_PAPER=false diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..01ce690 --- /dev/null +++ b/.gitignore @@ -0,0 +1,13 @@ +.venv/ +__pycache__/ +*.pyc + +.env + +bot_state.json +day_pnl.json +recent_sells.json +trades.jsonl +trading_bot.log + +*.log diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..d8eddea --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,96 @@ +# AGENTS.md — AI 助手项目上下文 + +> 本文件供 AI 编码助手（如 OpenCode）在新会话中快速恢复项目上下文。 +> **修改本项目后请同步更新本文件。** + +## 项目概述 + +Interactive Brokers 股票自动交易机器人（Python + ib_insync），多策略并行。 +**实盘运行**：账户 `U4845070`，IB Gateway 端口 4001（Gateway Live）。 +另有账户内独立持仓 `HF x300` —— 永远不属于 bot 管理，禁止触碰。 + +交易标的（1分钟K线，每策略每标的最多1笔）：**AMZN、AAPL、NVDA、NOK**（2026-08-03 起加 NVDA、NOK；此前为11只科技股的组合已停用）。 +每标的持仓价值上限 `max_symbol_value_usd`=$2300（三策略合计，`tracker.symbol_value()` 计算）。 + +## 文件结构与职责 + +| 文件 | 职责 | +|------|------| +| `config.py` | dataclass 配置；自动加载 `.env`（python-dotenv）；`state_file` 为基于 `__file__` 的绝对路径 | +| `connection.py` | IB 连接单例 `ib_conn`（connect/reconnect/ensure_connected） | +| `bars.py` | `BarManager`：每个合约一条 `keepUpToDate=True` 实时K线订阅，多策略共享；`to_completed_df` 丢弃未收盘bar；`is_market_active` 通过最新bar时效判断开闭市 | +| `state.py` | `PositionTracker`：持仓归属（哪个策略拥有哪笔仓位），持久化到 `bot_state.json`；启动/重连时 `reconcile()` 对账；每笔买卖追加写入 `trades.jsonl` 流水账本（首次运行以当前持仓为 seed） | +| `daily_report.py` / `report.sh` | 当日成交明细+盈亏报告（FIFO，数据源 `trades.jsonl`）；用法 `./report.sh [YYYY-MM-DD]` | +| `orders.py` | `execute_market_order`：下单+等待成交（30s超时撤单）+防重复单（`has_open_order`，按 orderRef 匹配） | +| `strategies/` | `ma_cross`(SMA20/50+ADX)、`short_term`(EMA5/10+VWAP+max_hold)、`mean_reversion`(RSI/布林/急跌抄底)、`forex`(默认关闭) | +| `main.py` | 主循环60s；disconnectedEvent 只注册一次且有并发/关机防护；重连后 `bar_manager.reset()` + 重新 `on_start` | + +## 关键设计决策（不要违反） + +1. **持仓归属**：IB 只报账户级持仓，归属靠 `bot_state.json`。策略**只卖自己的持仓**。账户中未被追踪的持仓标记为 UNMANAGED，永不出卖。 +2. **交易所硬止损（主）+ 软止损（兜底）**：买入成交后立即挂 GTC STP 卖单（`stop_loss_pct`，outsideRth=True），bot 停机/断线期间仍由交易所执行。`base.py` 止损单机制：每周期 `_sync_stop_orders()`（成交→record_sell；终态订单按 `_trade_uid`(permId/orderId) **只消费一次**；只认领未完成订单；owned 但无单→补挂）；策略卖出前 `_cancel_stop()` 并处理撤单竞态中的成交；软止损仅在 `_has_active_stop()==False` 时兜底。外汇仍用软止损（市场连续，且策略默认关闭）。 +3. **卖出不阻塞**：死叉类退出若盈利不足 `min_profit_pct`（1.5%），**每轮重检**（不是等下次交叉事件），达标即卖。 +4. **按金额定股数**：每笔 `trade_value_usd`（$2000，07-27 从 $1000 提高）按信号价折算（`base._order_quantity`，floor、最少1股）；佣金占比约 0.05%+0.1%往返。 +5. **全局持仓上限**：`config.max_positions`（12 批，≈$24k 敞口，07-28 从 15 调低）。策略买入前检查 `tracker.total_positions()`，达上限则跳过买入（日志 SKIPPED）；已有持仓正常管理退出，不强制平仓。 +6. **每日亏损熔断**：`config.max_daily_loss`（$150）。`PositionTracker` 按 record_sell 实时累计当日已实现盈亏（持久化 `day_pnl.json`，跨重启保留、本地午夜重置），触限后当日禁止开新仓（`base._can_open_position` 统一闸门，返回原因）。 +7. **止损/失败冷却**：同一策略同一标的被止损（硬止损成交或软止损卖出）后 `stop_cooldown_minutes`（30分钟）内禁止重买；下单被拒/超时也进冷却（防每周期重试刷屏）。 +8. **全局卖出冷却（跨策略防 flip-flop）**：任何策略卖出某标的后，所有策略在 `sell_cooldown_minutes`（30分钟）内禁止买入该标的；超过冷却但仍在 `sell_improvement_window_minutes`（60分钟）内，重买价必须 ≥ `sell_improvement_pct`（0.5%）低于最近卖价。登记点统一在 `tracker.record_sell` / `_record_external_sell`（覆盖策略卖出、硬止损成交、撤单竞态成交、断线回填），持久化到 `recent_sells.json`（跨重启保留，惰性清理）。2026-08-04 修复当日 3 起同价 flip：NOK 卖→买（跨策略）、NVDA 卖→买（同策略 4 分钟原价买回）、AAPL 卖→买（跨策略）。 +8. **MeanRev 趋势过滤**：仅在 `close > SMA(trend_ma_period=50)` 时允许抄底，避免下跌趋势中接飞刀。 +9. **信号只用已收盘K线**：最后一根 forming bar 必须丢弃（时间戳比较法）。 +10. **休市禁交易**：最新bar年龄 > `bar_seconds*5` 视为休市，跳过全部信号。收盘后约5分钟内 bar 仍"新鲜"，此窗口的市价单会隔夜排队——收市停机需提前（参考 15:59 EOD 停止的做法）。 +11. **K线流量**：禁止每轮全量拉历史数据（旧版 5 小时 845MB）；用 keepUpToDate 订阅（约 3.5MB/5小时）。`formatDate=2`（epoch，时区无歧义）。 + +## ⚠️ 已修复的 bug（勿重新引入） + +- **ADX**：`down_move = -low.diff()`（Wilder 定义），**不是** `low.diff().abs()`。abs 版本会把上涨中的低点抬升误记为空头 DM，导致趋势方向完全颠倒。 +- **RSI**：`rs = avg_gain / avg_loss` 直接除，让 `inf` 自然传播（纯涨→RSI 100）。不要 `.replace(0, nan)`，否则纯涨时 RSI 变 NaN。 +- **state_file 必须绝对路径**：cron/nohup 启动时 CWD 是 `$HOME`，相对路径会把状态文件写错位置。 +- **从 shell 工具启动 bot 必须 `setsid`**：`setsid nohup .venv/bin/python main.py >> trading_bot.log 2>&1 < /dev/null &`。否则终端会话结束/工具超时杀进程组时 bot 会被带走。 +- **BarManager.subscribe 竞态**：先存 in-flight Task 再 await（重连任务的 `on_start` 与主循环 `on_bar` 会并发订阅同一合约，旧代码 `await` 后才存导致重复订阅+订阅泄漏）。主循环断线期间直接跳过周期。 +- **断线窗口的止损成交必须回填账本**：STP 在 bot 断线时成交，bot 收不到事件；`reconcile()` 清理/裁剪仓位时必须调用 `_record_external_sell` 补记 `trades.jsonl`（价格取 `reqCompletedOrders` 实际成交价，取不到则按止损价 -2% 估记并标 `est:true`）。否则报告出现幽灵持仓、漏记亏损（07-30 修复，含一次性历史修复）。 +- 重连事件重复注册、关机后误触发重连：已在 main.py 修复，注意保持 `_running` 与 `_reconnect_task` 防护逻辑。 +- **`reqCompletedOrdersAsync` 必须传 `apiOnly` 参数**：ib_insync 0.9.86 起签名必填。bot 回填断线 STP 成交用 `apiOnly=True`（只取 API 订单，天然排除 TWS 手工单）。不传会抛 TypeError，断线回填静默失效（07-31 修复）。 +- **`reqCompletedOrdersAsync` 会把已完成订单注入 `wrapper.trades` 造成"幻影订单"**：ib_insync 0.9.86 的 `wrapper.completedOrder`（wrapper.py:402）以非终态 OrderStatus（filled=0、status 取 orderState.status）把全部已完成订单塞进 `ib.trades()`/`ib.openTrades()`——同 ref 的僵尸单（如 08-03 遗留 NOK STP）在重启后以 `PreSubmitted/done=False` 现形 → `has_open_order` 永远返回 True → 该 key 的止损**永不补挂**（持仓裸奔）；done 幻影还会触发每轮 "ended with status=Filled - re-placing now" 噪音。防御（08-04 修复）：`main._completed_stp_fills` 收集完成交价后按 `permId` 把幻影从 `ib.wrapper.trades` 弹出；`orders.has_open_order` 另加 `remaining > 0` 硬条件（幻影 remaining 恒为 0，真实在挂单 >0）。 +- **外部成交价 0.0 必须按"未知"处理**：`reqCompletedOrders` 对跨会话（上次 bot 会话）的已完成订单可能返回 `avgFillPrice=0.0`。`main.py` 的 `_completed_stp_fills` 要跳过 `fill_price <= 0` 的订单；`state.py` 的 `_record_external_sell` 用 `estimated = not price`（None 或 0.0 都视为取不到）→ 按止损价 -2% 估记并标 `est:true`。否则会把 0.0 当真价写账本，亏损虚算（07-31 修复：迁移目录时踩中，造成 day_pnl 虚增 -1668.74）。 +- **僵尸 STP（PreSubmitted 永不变更）+ `has_open_order` 造成"假卡死"**：被风控拒掉/IB 端失同步的 STP 会以 PreSubmitted 卡在 `ib.openTrades()` 里（isDone()=False），`has_open_order` 永远返回 True → `_place_stop` 静默 skip，且 `_raise_stop_if_needed` 只在 target>current 时撤单重挂，价格不符的僵尸单永不处理 → bot 看起来"卡死"（无日志，持仓无有效 STP）。症状：`reqAllOpenOrdersAsync`（跨 clientId）可见这些单，`cancelOrder` 报 Error 10147 取不掉。恢复：`reqGlobalCancel()` 清掉全部僵尸单后重启。防御（08-03 修复）：`_raise_stop_if_needed` 改为 **auxPrice 与目标价不匹配（容差 <0.01）就撤单重挂**，不只处理上移。另：`has_open_order` 只看 `openTrades()`（本 clientId 自己的单），`reqOpenOrdersAsync` 返回本 clientId 的订单，诊断全账户订单要用 `reqAllOpenOrdersAsync`。 +- **Cancelled STP 必须当轮重挂**：`_sync_stop_orders` 消费终态订单后若 owned 且未成交，**同一周期**就 `_place_stop`，不要拖到下轮（旧代码 `continue` 后依赖下一轮 by_ref 找不到已 consumed 的单才重挂，重启后首轮只挂 1 条就停）。08-03 修复。 +- **正式目录是 `~/Desktop/DeepSeek/BotDeepSeek`**：bot 自此从该目录启动（`restart_bot.sh` 自定位）。`~/Desktop/kimi/ib-trading-bot-kimi` 仅作备份保留，勿再从那里启动；两目录代码保持一致。 + +## 环境特殊性 + +- `.venv` **无 pip/ensurepip**；装包用 `python3 -m pip install --target=.venv/lib/python3.12/site-packages ` +- pandas 3.0.3：避免使用已移除的 API（如 groupby.apply 的 include_groups）；VWAP 用 groupby+cumsum 向量化实现 +- 系统无 `at`，有一次性任务用 **cron + 执行后自删**（`crontab -l | grep -v | crontab -`），参考 `close_legacy_positions.sh` / `stop_bot_eod.sh` +- 系统时区 EDT = 美股时间；cron 按本地时间 +- IB Gateway 自带 JRE 17（`~/.local/share/i4j_jres/`）；`~/jdk`（Temurin 21）是备用，`start_gateway.sh` 引用它 +- Gateway API：clientId=1 给 bot；临时只读检查用 clientId=77，查完即断开 + +## 安全红线 + +- **这是实盘账户**：任何测试不得真实下单。验证用离线测试（纯函数：指标/状态/K线工具），参考做法：构造合成数据测 `_calc_rsi/_calc_adx/_calc_vwap`、`PositionTracker`、`to_completed_df` +- 不要为了"测试"运行 `main.py`——它会下真实订单 +- 修改策略逻辑后：先 `py_compile`，再跑离线测试，最后才考虑重启 bot + +## 日常操作 + +```bash +./start_gateway.sh # 启动 IB Gateway（GUI，需登录状态） +./restart_bot.sh # 重启 bot（杀旧进程 + nohup 启动） +tail -f trading_bot.log # 运行日志 +cat bot_state.json # 当前策略持仓归属与成本 +``` + +一次性定时任务示例（cron）：`close_legacy_positions.sh`（平仓+自动启动bot）、`stop_bot_eod.sh`（15:59 收市前停机）。 + +## 当前状态（2026-08-03 盘中） + +- **已恢复正常交易**（`sell_only=False`），标的 AMZN/AAPL/NVDA/NOK，每标的持仓价值上限 $2300 +- 当前持仓（08-03 盘中）：ShortTerm AMZN x7 @285.86、NVDA x9 @208.17、NOK x212 @9.3997；MAStock AAPL x6 @304.08。四笔均有 GTC STP 在交易所（clientId=1） +- 交易参数：每笔 ~$2000 折股、min_profit 1.5%、入场 2 根K线确认、止损 -2.5% STP（2026-08-03 从 -2% 放宽，依据 256 笔成交统计：61 笔亏损平仓中 82% 已控制在 -3% 内，2% 执行良好；放宽仅为减少 whipsaw，每笔风险 +$10 仍在 $150 日熔断内）、每日熔断 $150、全局持仓 12 批 +- **移动止损方案A**（08-03 实现）：盈利 ≥ min_profit(1.5%) 后，STP 从"入场价-2.5%"上移到"持仓期间最高收盘价-2.0%"，且永不低于固定止损。`use_trailing_stop/trailing_stop_pct` 在 config 三股票策略里 +- **持仓不设上限（08-05 撤销 5 天强制卖出）**：`ShortTermConfig.max_hold_days=0`（0=不限），short_term.py 仅在 `>0` 时触发 max-hold 卖出。持仓由 EMA 死叉+min_profit、硬止损、移动止损自然退出，可长期持有趋势单。 +- **全局卖出冷却（08-04 实现，方案C）**：`sell_cooldown_minutes=30`（任何卖出后所有策略禁买）+ `sell_improvement_pct=0.5%`（冷却后**120分钟内**重买须比最近卖价低0.5%，08-06 由60→120）+ 登记在 `recent_sells.json` +- **硬止损成交当日禁买（08-06）**：`_mark_hard_stop_cooldown` 使硬止损成交后该标的冷却到**次日零时**（`_next_day_start`），防隔夜跳空止损后当天同价回补；软止损/订单失败冷却仍 30 分钟。 +- **NOK 减半仓位（08-06）**：`config.symbol_trade_value_usd={"NOK": 1000.0}`，base `_trade_value_usd` 供各策略 `_order_quantity/_can_open_position` 使用，其余标的一律默认 2000。 +- **MAStock 入场过滤收紧（08-06）**：ADX 门槛 25→30（`adx_min`），且新增 `require_slow_ma_slope=True`——仅当慢 MA 自身在确认窗口内上升才允许买入（防下跌趋势中金叉接刀，08-05 NOK 案例）。 +- **过夜敞口观察中（2026-07-31 起）**：07-31 早盘 AAPL x5 隔夜跳空 -8.9% 止损成交（STP @265.52→实际 304.145，亏损 $148≈止损预期的4倍）。决策暂不改为 EOD 平仓，先积累 AMZN/AAPL 新 regime 样本（2-4周）统计跳空频率/实际过夜损耗，再决定是否加 EOD 强制平仓。 diff --git a/README.md b/README.md new file mode 100644 index 0000000..52269e7 --- /dev/null +++ b/README.md @@ -0,0 +1,141 @@ +# Interactive Brokers 自动交易系统 + +基于 Python + ib_insync 的 IB 自动化交易框架，支持多策略并行、持仓归属追踪、 +自动止损、断线自动重连。 + +## 项目结构 + +``` +ib-trading-bot-kimi/ +├── config.py # 配置管理（dataclass + .env 环境变量） +├── connection.py # TWS/IB Gateway 连接管理（含自动重连） +├── bars.py # 共享K线订阅管理（keepUpToDate 增量推送） +├── orders.py # 下单执行（成交确认、超时撤单、防重复单） +├── state.py # 持仓归属追踪（JSON 持久化 + 启动对账） +├── main.py # 主程序入口 +├── strategies/ +│ ├── __init__.py +│ ├── base.py # 策略基类 +│ ├── ma_cross.py # SMA 20/50 金叉 + ADX 趋势过滤 +│ ├── short_term.py # EMA 5/10 金叉 + VWAP 确认 + 最长持仓限制 +│ ├── mean_reversion.py # RSI超卖/布林带下轨/急跌 抄底 +│ └── forex.py # 外汇均线交叉（默认关闭） +├── bot_state.json # 运行时生成：持仓归属状态（勿手动编辑，除非迁移持仓） +├── requirements.txt +└── .env.example +``` + +## 策略说明 + +| 策略 | 入场 | 出场（每轮重检） | +|------|------|------------------| +| MAStockStrategy | 金叉 + 连续2根确认 且 ADX>25 | **交易所STP止损 -2%**；快线跌破慢线且盈利 ≥1.5% | +| ShortTermMAVWAP | EMA金叉 + 连续2根确认 且 价格>VWAP | **交易所STP止损 -2%**；持仓≥5天；死叉且盈利 ≥1.5% | +| MeanReversion | RSI<30 / 触及布林下轨（均需连续2根）/ 5根急跌1% | **交易所STP止损 -2%**；RSI>70 / 回到布林中轨 / 反弹1.5%（需盈利≥1.5%） | +| ForexMAStrategy | 快线金叉慢线 | **软止损 -1%**；死叉 | + +**交易规模**：每笔按 `trade_value_usd`（默认 $2000）按信号价折算股数 +（如 $600 的股买 3 股，$95 的股买 21 股），每策略每标的最多一笔。 +**入场确认**：交叉类信号要求交叉后下一根 K 线仍同向，超卖类信号要求 +连续 2 根 K 线满足条件，过滤单根毛刺、降低交易频率。 + +所有策略共用 2 只标的（AMZN、AAPL），但**各自独立管理自己的持仓**（见下文"持仓归属"）。 + +## 风控机制 + +1. **交易所硬止损（主）+ 软止损（兜底）**：每笔买入成交后，立即在交易所挂 + **GTC 止损卖单（STP）**（`stop_loss_pct`，股票默认 -2%），`outsideRth=True` 盘后/隔夜 + 同样有效——即使 bot 停止、断网、电脑关机，止损仍由交易所执行。bot 每 60 秒同步 + 止损单状态：成交→更新持仓归属；被撤/丢失→自动补挂；重启→自动认领。策略正常卖出前 + 会先撤掉对应止损单。若止损单因故不存在（如挂单被拒），原"软止损"（每 60 秒检查 + 收盘价）自动兜底。 + ⚠️ STP 触发后变成市价单，快速行情中成交价可能略低于止损价（滑点）。 +2. **持仓归属**：`bot_state.json` 记录每笔持仓属于哪个策略。策略只卖自己的持仓， + 不会互相平仓。启动/重连时自动与账户真实持仓对账： + - 状态中有、账户没有 → 清除（说明已被手动平仓或断线期间成交） + - 账户中有、状态没有 → 标记为 **UNMANAGED**（bot 永不会卖，需手动处理或迁移） +3. **最低盈利限制解除阻塞**：死叉卖出被盈利条件阻止后，每轮都会重检， + 盈利达标即卖；同时止损兜底，不会无限套牢。 +3b. **全局持仓上限**：`max_positions`（默认 12 批）限制所有策略的并发持仓总数， + 控制最大敞口约 $24k（12 × $2000）。达上限后新买入被跳过（日志记录 SKIPPED）， + 已有持仓正常管理退出，不会被强制平仓。 +3c. **每日亏损熔断**：当日已实现亏损达到 `max_daily_loss`（默认 $150）后， + 当天禁止开新仓（次日自动重置，已有持仓正常退出）。 +3d. **止损冷却期**：同一策略同一标的被止损后 30 分钟内禁止重买， + 避免下跌通道中"买入→止损→再买"的连续失血；下单被拒/超时同样进入冷却。 +3e. **MeanRev 趋势过滤**：仅在收盘价高于 SMA50 时允许抄底，不接下跌中的飞刀。 +4. **完整K线信号**：信号只用已收盘的K线计算，避免未收完K线导致的信号闪烁。 +5. **休市禁交易**：通过最新K线时间判断市场是否开盘，收盘/周末/节假日自动暂停信号。 +6. **订单确认**：下单后等待成交（默认30秒），超时自动撤单；同一策略同一标的 + 有未成交单时不会重复下单。每笔订单带 `orderRef`（如 `MAStockStrategy:META`）方便在 TWS 中识别来源。 +7. **增量K线订阅**：每个标的只订阅一次 `keepUpToDate` 实时K线流，三个策略共享， + 替代原来每分钟全量拉取（原来 5 小时约 845MB 流量）。 + +## 配置 + +复制 `.env.example` 为 `.env` 后编辑（程序自动加载 `.env`）： + +| 变量 | 默认值 | 说明 | +|------|--------|------| +| `IB_HOST` | `127.0.0.1` | IB Gateway 地址 | +| `IB_PORT` | `4001` | **4001=Gateway实盘**, 4002=Gateway模拟, 7496=TWS实盘, 7497=TWS模拟 | +| `IB_CLIENT_ID` | `1` | 客户端唯一 ID | +| `IB_ACCOUNT` | `U4845070` | 指定下单账户 | +| `IB_PAPER` | `false` | 仅用于启动日志警示；真实模式由端口/Gateway登录类型决定 | + +策略参数在 `config.py` 中调整：`symbols`、`fast_ma_period`、`slow_ma_period`、 +`trade_quantity`、`stop_loss_pct`、`min_profit_pct` 等。 + +## 运行 + +```bash +# 1. 启动 IB Gateway（脚本会自动准备 JDK；注意 /tmp 重启后会被清空） +./start_gateway.sh +# 在 Gateway 中登录，并确认 API 设置：Enable ActiveX and Socket Clients、 +# 端口、Trusted IPs 加入 127.0.0.1 + +# 2. 启动 bot（前台调试） +.venv/bin/python main.py + +# 或后台重启（会自动先杀掉旧进程） +./restart_bot.sh +tail -f trading_bot.log + +# 3. 查看当天成交明细与盈亏（可指定日期） +./report.sh [YYYY-MM-DD] +``` + +启动时如为实盘模式，日志会有醒目的 `LIVE TRADING MODE` 警示。 + +## ⚠️ 从旧版本迁移（重要） + +旧版本没有持仓归属记录。如果升级时账户里还有旧 bot 开的仓位，新 bot +会把它们标记为 **UNMANAGED 并永远不管**（不会止损、不会止盈）。两个选择： + +1. **手动平掉旧仓位**后再启动新 bot（推荐，最干净）； +2. 或手动编辑 `bot_state.json` 把旧仓位归属到某个策略，例如： + +```json +{ + "MAStockStrategy": { + "NVDA": {"quantity": 1, "entry_price": 207.33, "entry_date": "2026-07-20"} + } +} +``` + +注意：归属后该仓位立即受止损/止盈规则约束，若浮亏已超过 `stop_loss_pct`， +启动后第一轮检查就会被止损卖出。 + +### 一次性迁移任务（2026-07-22 已排期） + +`close_legacy_positions.py` + cron 实现无人值守迁移：开盘后（09:45 ET） +自动卖出 5 笔旧仓位（NVDA/META/AVGO/MSFT/PLTR 各 1 股），**成功后才自动 +启动新 bot**；失败则不启动（避免遗留 UNMANAGED 仓位），日志见 +`close_legacy_positions.log`。该 cron 任务执行后自动删除。 + +## 安全警告 + +- 当前配置为**实盘模式**（4001 端口），下单即真实成交 +- 策略有止损，但软止损依赖程序持续运行；请保证机器、网络、Gateway 稳定 +- 建议定期查看 `trading_bot.log` 和 `bot_state.json` +- 三个策略对同一标的最多各持 1 份（共 3 份），请确认账户风险敞口可接受 diff --git a/bars.py b/bars.py new file mode 100644 index 0000000..89061a2 --- /dev/null +++ b/bars.py @@ -0,0 +1,127 @@ +import asyncio +import logging +import re +from datetime import datetime, timedelta, timezone + +import pandas as pd +from ib_insync import IB, Contract + +logger = logging.getLogger(__name__) + +_UNIT_SECONDS = { + "sec": 1, + "min": 60, + "hour": 3600, + "day": 86400, + "week": 604800, + "month": 2592000, +} + +def parse_bar_size_seconds(bar_size: str) -> int: + """Parse IB bar size strings like '1 min', '5 secs', '1 hour' to seconds.""" + m = re.match(r"^\s*(\d+)\s*([a-zA-Z]+)\s*$", bar_size) + if not m: + raise ValueError(f"Unsupported bar size: {bar_size!r}") + n, unit = int(m.group(1)), m.group(2).lower() + for prefix, seconds in _UNIT_SECONDS.items(): + if unit.startswith(prefix): + return n * seconds + raise ValueError(f"Unsupported bar size unit: {bar_size!r}") + +class BarManager: + """Maintains one live (keepUpToDate) historical bar subscription per + (contract, bar size, data type), shared by all strategies. + + This replaces per-cycle full history downloads: after the initial fetch, + IB pushes incremental updates, cutting traffic dramatically. + """ + + def __init__(self, ib: IB): + self.ib = ib + # key -> BarDataList (ready) or asyncio.Task (subscription in flight). + # Storing the in-flight task makes concurrent subscribe() calls for the + # same key share one request instead of firing duplicate subscriptions. + self._bars: dict[tuple, object] = {} + + @staticmethod + def _key(contract: Contract, bar_size: str, what_to_show: str) -> tuple: + return (contract.conId, bar_size, what_to_show) + + async def subscribe(self, contract: Contract, bar_size: str, duration: str, + what_to_show: str, use_rth: bool = True): + + key = self._key(contract, bar_size, what_to_show) + existing = self._bars.get(key) + if existing is not None: + if isinstance(existing, asyncio.Task): + return await existing # share the in-flight request + return existing + + task = asyncio.create_task(self.ib.reqHistoricalDataAsync( + contract=contract, + endDateTime="", + durationStr=duration, + barSizeSetting=bar_size, + whatToShow=what_to_show, + useRTH=use_rth, + formatDate=2, # epoch seconds -> unambiguous UTC timestamps + keepUpToDate=True, + )) + self._bars[key] = task + try: + bars = await task + except Exception: + self._bars.pop(key, None) # allow retry on failure + raise + self._bars[key] = bars + logger.info( + "Bar subscription started: conId=%s %s %s (%d bars)", + contract.conId, bar_size, what_to_show, len(bars), + ) + return bars + + def get(self, contract: Contract, bar_size: str, what_to_show: str): + + v = self._bars.get(self._key(contract, bar_size, what_to_show)) + return None if isinstance(v, asyncio.Task) else v + + def reset(self): + """Cancel all live subscriptions (e.g. after a disconnect).""" + + for key, v in self._bars.items(): + try: + if isinstance(v, asyncio.Task): + v.cancel() + else: + self.ib.cancelHistoricalData(v) + except Exception: + pass + if self._bars: + logger.info("Bar subscriptions cleared (%d)", len(self._bars)) + self._bars.clear() + +def to_completed_df(bars, bar_seconds: int) -> pd.DataFrame: + """Convert bars to a DataFrame containing only completed bars. + + The last bar of a live subscription is still forming; signals computed on + it would repaint, so it is dropped until it closes. + """ + if not bars: + return pd.DataFrame() + df = pd.DataFrame(bars) + df["date"] = pd.to_datetime(df["date"], utc=True) + now = datetime.now(timezone.utc) + if len(df) and df["date"].iloc[-1] + timedelta(seconds=bar_seconds) > now: + df = df.iloc[:-1] + return df + +def last_bar_age_seconds(df: pd.DataFrame) -> float: + """Age of the most recent completed bar. Large age => market closed/stale.""" + if df.empty: + return float("inf") + return (datetime.now(timezone.utc) - df["date"].iloc[-1]).total_seconds() + +def is_market_active(df: pd.DataFrame, bar_seconds: int, factor: int = 5) -> bool: + """Data-driven market-hours check: if the latest completed bar is too old, + the market is closed (overnight/weekend/holiday) and we must not trade.""" + return last_bar_age_seconds(df) <= bar_seconds * factor diff --git a/close_legacy_positions.py b/close_legacy_positions.py new file mode 100644 index 0000000..7420a79 --- /dev/null +++ b/close_legacy_positions.py @@ -0,0 +1,120 @@ +"""One-shot task: close the legacy (pre-refactor) positions after market open. + +Sells up to LEGACY_QTY shares of each legacy symbol with market orders, +waits for fill confirmation, then verifies the account is flat for them. + +Scheduled via cron for 2026-07-22 09:45 ET; also safe to run manually: + .venv/bin/python close_legacy_positions.py +""" +import asyncio +import logging +import sys +from datetime import datetime, timedelta + +from config import config +from connection import ib_conn +from orders import create_stock_contract, execute_market_order + +# legacy positions observed in trading_bot.log on 2026-07-21 +LEGACY_QTY = {"NVDA": 1, "META": 1, "AVGO": 1, "MSFT": 1, "PLTR": 1} + +CONNECT_RETRY_SECONDS = 30 +CONNECT_DEADLINE_MINUTES = 60 # gateway might be started a bit late +ORDER_TIMEOUT = 60 + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", +) +logging.getLogger("ib_insync.wrapper").setLevel(logging.WARNING) +logging.getLogger("ib_insync.client").setLevel(logging.WARNING) +logger = logging.getLogger("close_legacy") + + +async def connect_with_retries() -> bool: + deadline = datetime.now() + timedelta(minutes=CONNECT_DEADLINE_MINUTES) + attempt = 0 + while datetime.now() < deadline: + attempt += 1 + if await ib_conn.ensure_connected(): + logger.info("Connected on attempt %d", attempt) + return True + logger.warning( + "Connect attempt %d failed, retrying in %ds (giving up at %s)...", + attempt, CONNECT_RETRY_SECONDS, deadline.strftime("%H:%M"), + ) + await asyncio.sleep(CONNECT_RETRY_SECONDS) + return False + + +async def main() -> int: + logger.warning("=" * 60) + logger.warning("LEGACY POSITION CLEANUP - LIVE account %s", config.ib.account) + logger.warning("Will market-SELL up to: %s", LEGACY_QTY) + logger.warning("=" * 60) + + if not await connect_with_retries(): + logger.error( + "Could not connect to IB Gateway within %d minutes - aborting. " + "Is the Gateway running and logged in?", + CONNECT_DEADLINE_MINUTES, + ) + return 1 + + ib = ib_conn.ib + accounts = ib.managedAccounts() + logger.info("Managed accounts: %s", accounts) + if config.ib.account and config.ib.account not in accounts: + logger.error( + "Configured account %s not in managed accounts %s - aborting", + config.ib.account, accounts, + ) + ib_conn.disconnect() + return 1 + + positions: dict[str, float] = {} + for p in ib.positions(): + if p.contract.secType == "STK": + positions[p.contract.symbol] = positions.get(p.contract.symbol, 0) + p.position + logger.info("Current stock positions: %s", positions) + + failures = 0 + for symbol, expected_qty in LEGACY_QTY.items(): + actual = positions.get(symbol, 0) + sell_qty = min(actual, expected_qty) + if sell_qty <= 0: + logger.info("%s: no legacy position (actual=%g), skipping", symbol, actual) + continue + contract = create_stock_contract(symbol) + await ib.qualifyContractsAsync(contract) + trade = await execute_market_order( + ib, contract, "SELL", sell_qty, f"LegacyCleanup:{symbol}", timeout=ORDER_TIMEOUT + ) + if trade and trade.orderStatus.filled > 0: + logger.info( + "%s: SOLD %g @ %.2f", symbol, + trade.orderStatus.filled, trade.orderStatus.avgFillPrice, + ) + else: + logger.error("%s: SELL FAILED", symbol) + failures += 1 + + # verify the account is flat for the legacy symbols + await asyncio.sleep(2) + remaining = { + p.contract.symbol: p.position + for p in ib.positions() + if p.contract.secType == "STK" and p.contract.symbol in LEGACY_QTY and p.position > 0 + } + if remaining: + logger.error("Cleanup incomplete, remaining positions: %s", remaining) + else: + logger.info("All legacy positions closed.") + + ib_conn.disconnect() + return 1 if (failures or remaining) else 0 + + +if __name__ == "__main__": + sys.exit(asyncio.run(main())) diff --git a/close_legacy_positions.sh b/close_legacy_positions.sh new file mode 100755 index 0000000..4885111 --- /dev/null +++ b/close_legacy_positions.sh @@ -0,0 +1,33 @@ +#!/bin/bash +# One-shot scheduled task: close legacy positions, then start the new trading bot. +# Removes its own cron entry after running. +BOT_DIR="$(cd "$(dirname "$0")" && pwd)" +LOG="$BOT_DIR/close_legacy_positions.log" +BOT_LOG="$BOT_DIR/trading_bot.log" + +if pgrep -f "python.*main\.py" > /dev/null; then + echo "$(date) ABORT: trading bot is already running; skipping cleanup" >> "$LOG" +else + echo "=== $(date) legacy cleanup start ===" >> "$LOG" + "$BOT_DIR/.venv/bin/python" "$BOT_DIR/close_legacy_positions.py" >> "$LOG" 2>&1 + RC=$? + echo "=== $(date) legacy cleanup done (exit=$RC) ===" >> "$LOG" + + if [ "$RC" -eq 0 ]; then + echo "=== $(date) cleanup OK, starting new trading bot ===" >> "$LOG" + setsid nohup "$BOT_DIR/.venv/bin/python" "$BOT_DIR/main.py" >> "$BOT_LOG" 2>&1 < /dev/null & + BOT_PID=$! + sleep 8 + if kill -0 "$BOT_PID" 2>/dev/null; then + echo "=== bot started successfully (PID: $BOT_PID) ===" >> "$LOG" + else + echo "=== bot FAILED to start, check $BOT_LOG ===" >> "$LOG" + fi + else + echo "=== $(date) cleanup FAILED (exit=$RC) - bot NOT started." >> "$LOG" + echo "=== Fix manually (run: .venv/bin/python close_legacy_positions.py), then ./restart_bot.sh ===" >> "$LOG" + fi +fi + +# self-remove from crontab (one-shot task) +crontab -l 2>/dev/null | grep -v "close_legacy_positions" | crontab - diff --git a/config.py b/config.py new file mode 100644 index 0000000..fdec04a --- /dev/null +++ b/config.py @@ -0,0 +1,145 @@ +import os +from dataclasses import dataclass, field +from pathlib import Path +from typing import Optional + +_PROJECT_DIR = Path(__file__).resolve().parent + +try: + from dotenv import load_dotenv + load_dotenv() +except ImportError: + pass + + +@dataclass +class IBConfig: + host: str = os.getenv("IB_HOST", "127.0.0.1") + port: int = int(os.getenv("IB_PORT", "4001")) + client_id: int = int(os.getenv("IB_CLIENT_ID", "1")) + account: Optional[str] = os.getenv("IB_ACCOUNT", "U4845070") + is_paper: bool = os.getenv("IB_PAPER", "false").lower() == "true" + + @property + def port_label(self) -> str: + if self.port == 4002: + return "Gateway Paper (4002)" + elif self.port == 4001: + return "Gateway Live (4001)" + elif self.port == 7497: + return "TWS Paper (7497)" + elif self.port == 7496: + return "TWS Live (7496)" + return f"Unknown ({self.port})" + + +@dataclass +class StockStrategyConfig: + enabled: bool = True + symbols: list[str] = field(default_factory=lambda: ["AMZN", "AAPL", "NVDA", "NOK"]) + currency: str = "USD" + exchange: str = "SMART" + bar_size: str = "1 min" + lookback_days: int = 5 + fast_ma_period: int = 20 + slow_ma_period: int = 50 + trade_value_usd: float = 2000.0 # per-trade dollar amount; qty = floor(value / price) + min_profit_pct: float = 1.5 + stop_loss_pct: float = 2.5 + use_hard_stop: bool = True # place real GTC STP orders at the exchange + entry_confirm_bars: int = 2 # entry signal must hold for N consecutive bars + stop_cooldown_minutes: int = 30 # no re-entry on same symbol for N min after a stop-out + use_trailing_stop: bool = True # once profit >= min_profit, trail the stop below the peak high + trailing_stop_pct: float = 2.0 # trailing stop distance below peak high + adx_min: float = 30.0 # ADX trend filter for entries (was 25; 08-06 raised to avoid weak-trend whipsaw) + require_slow_ma_slope: bool = True # only enter if the slow MA is itself rising (no knife-catching in a downtrend) high + + +@dataclass +class ShortTermConfig: + enabled: bool = True + symbols: list[str] = field(default_factory=lambda: ["AMZN", "AAPL", "NVDA", "NOK"]) + currency: str = "USD" + exchange: str = "SMART" + bar_size: str = "1 min" + lookback_days: int = 5 + fast_ma_period: int = 5 + slow_ma_period: int = 10 + trade_value_usd: float = 2000.0 + max_hold_days: int = 0 # 0 = unlimited hold (08-05 撤销 5 天上限) + min_profit_pct: float = 1.5 + stop_loss_pct: float = 2.5 + use_hard_stop: bool = True + entry_confirm_bars: int = 2 + stop_cooldown_minutes: int = 30 + use_trailing_stop: bool = True + trailing_stop_pct: float = 2.0 + + +@dataclass +class ForexStrategyConfig: + enabled: bool = False + pairs: list[str] = field(default_factory=lambda: ["EUR.USD", "GBP.USD"]) + exchange: str = "IDEALPRO" + bar_size: str = "1 min" + lookback_days: int = 5 + fast_ma_period: int = 12 + slow_ma_period: int = 26 + trade_units: int = 3000 + stop_loss_pct: float = 1.0 + + +@dataclass +class MeanReversionConfig: + enabled: bool = True + symbols: list[str] = field(default_factory=lambda: ["AMZN", "AAPL", "NVDA", "NOK"]) + currency: str = "USD" + exchange: str = "SMART" + bar_size: str = "1 min" + lookback_days: int = 5 + trade_value_usd: float = 2000.0 + rsi_period: int = 14 + rsi_oversold: float = 30.0 + rsi_overbought: float = 70.0 + bb_period: int = 20 + bb_std: float = 2.0 + rapid_drop_pct: float = 1.0 + rapid_drop_bars: int = 5 + recovery_pct: float = 1.5 + min_profit_pct: float = 1.5 + stop_loss_pct: float = 2.5 + use_hard_stop: bool = True + entry_confirm_bars: int = 2 + stop_cooldown_minutes: int = 30 + use_trailing_stop: bool = True + trailing_stop_pct: float = 2.0 + trend_ma_period: int = 50 # mean-reversion buys only above this SMA + + +@dataclass +class AppConfig: + ib: IBConfig = field(default_factory=IBConfig) + stock: StockStrategyConfig = field(default_factory=StockStrategyConfig) + short_term: ShortTermConfig = field(default_factory=ShortTermConfig) + forex: ForexStrategyConfig = field(default_factory=ForexStrategyConfig) + mean_reversion: MeanReversionConfig = field(default_factory=MeanReversionConfig) + max_retries: int = 5 + retry_delay: int = 5 + loop_interval: float = 60.0 + order_timeout: float = 30.0 + max_positions: int = 12 # global cap on concurrent open lots across all strategies + max_daily_loss: float = 150.0 # stop opening new positions for the day beyond this realized loss + sell_only: bool = False # liquidation mode: no new buys, strategies only manage exits + max_symbol_value_usd: float = 2300.0 # per-symbol total position value cap (all strategies combined) + sell_cooldown_minutes: int = 30 # global: no strategy may (re)buy a symbol within N min of ANY sell + sell_improvement_pct: float = 0.5 # beyond the cooldown, re-buy only if price is >=0.5% below last sell + sell_improvement_window_minutes: int = 120 # how long the price-improvement rule applies after a sell + symbol_trade_value_usd: dict[str, float] = field( + # per-symbol per-trade dollar value override (halved for gap-prone NOK, 08-06) + default_factory=lambda: {"NOK": 1000.0} + ) + # absolute path: independent of the working directory the bot is started from + state_file: str = str(_PROJECT_DIR / "bot_state.json") + + +config = AppConfig() diff --git a/connection.py b/connection.py new file mode 100644 index 0000000..8749697 --- /dev/null +++ b/connection.py @@ -0,0 +1,56 @@ +import logging +from typing import Optional + +from ib_insync import IB + +from config import config + +logger = logging.getLogger(__name__) + + +class IBConnection: + def __init__(self): + self.ib = IB() + self._connected = False + + async def connect(self) -> bool: + if self._connected: + return True + try: + await self.ib.connectAsync( + host=config.ib.host, + port=config.ib.port, + clientId=config.ib.client_id, + account=config.ib.account or "", + ) + self._connected = True + logger.info( + "Connected to IB Gateway at %s:%s (%s)", + config.ib.host, config.ib.port, config.ib.port_label, + ) + return True + except Exception as e: + logger.error("Connection failed: %s", e) + self._connected = False + return False + + def disconnect(self): + if self._connected: + self.ib.disconnect() + self._connected = False + logger.info("Disconnected") + + async def reconnect(self) -> bool: + self.disconnect() + return await self.connect() + + def is_connected(self) -> bool: + return self._connected and self.ib.isConnected() + + async def ensure_connected(self) -> bool: + if not self.is_connected(): + return await self.reconnect() + return True + + +ib_conn = IBConnection() diff --git a/daily_report.py b/daily_report.py new file mode 100644 index 0000000..3dde57b --- /dev/null +++ b/daily_report.py @@ -0,0 +1,151 @@ +#!/usr/bin/env python +"""Daily trade report: today's fills + realized P&L from the trade ledger. + +Usage: + .venv/bin/python daily_report.py [YYYY-MM-DD] + ./report.sh [YYYY-MM-DD] + +Data source: trades.jsonl (append-only ledger written by PositionTracker). +P&L method: FIFO per (strategy, symbol). Current open lots shown at the end; +if the IB Gateway is reachable, live market values are included (read-only). +""" +import json +import os +import sys +from collections import defaultdict, deque +from datetime import date +from pathlib import Path + +LEDGER = Path(os.environ.get("TRADES_LEDGER", Path(__file__).resolve().parent / "trades.jsonl")) + +STRATEGY_SHORT = { + "MAStockStrategy": "MAStock", + "ShortTermMAVWAPStrategy": "ShortTerm", + "MeanReversionStrategy": "MeanRev", + "ForexMAStrategy": "Forex", +} + + +def short(name: str) -> str: + return STRATEGY_SHORT.get(name, name[:12]) + + +def load_and_replay(): + """Replay the whole ledger; return (today fills, realized maps, open lots).""" + lots = defaultdict(deque) # (strategy, symbol) -> deque([qty, price]) + fills_today = [] # display rows for the requested day + realized_sym = defaultdict(float) + realized_strat = defaultdict(float) + unknown_cost = 0.0 + day = sys.argv[1] if len(sys.argv) > 1 else date.today().isoformat() + + for line in LEDGER.read_text().splitlines(): + if not line.strip(): + continue + r = json.loads(line) + key = (r["strategy"], r["symbol"]) + ts_day, ts_time = r["ts"][:10], r["ts"][11:19] + + if r["type"] in ("seed", "buy"): + lots[key].append([r["qty"], r["price"]]) + if r["type"] == "buy" and ts_day == day: + fills_today.append((ts_time, r["strategy"], r["symbol"], "BUY", r["qty"], r["price"], None)) + + elif r["type"] in ("sell", "sell_external"): + pnl = 0.0 + remain = r["qty"] + dq = lots[key] + while remain > 1e-9 and dq: + take = min(dq[0][0], remain) + pnl += (r["price"] - dq[0][1]) * take + dq[0][0] -= take + remain -= take + if dq[0][0] <= 1e-9: + dq.popleft() + if remain > 1e-9: + pnl = None # cost basis unknown (lot predates the ledger) + if ts_day == day: + action = "SELL*" if r.get("est") else "SELL" + fills_today.append((ts_time, r["strategy"], r["symbol"], action, r["qty"], r["price"], pnl)) + if pnl is not None: + realized_sym[r["symbol"]] += pnl + realized_strat[r["strategy"]] += pnl + else: + unknown_cost += r["qty"] + + return day, fills_today, realized_sym, realized_strat, lots, unknown_cost + + +def live_positions(): + """Read-only: current positions + market values from IB. None if unreachable.""" + try: + import asyncio + from ib_insync import IB + from config import config as cfg + + async def _fetch(): + ib = IB() + await ib.connectAsync(cfg.ib.host, cfg.ib.port, clientId=77, timeout=10) + data = { + item.contract.symbol: (item.position, item.marketPrice, item.unrealizedPNL) + for item in ib.portfolio() + if item.contract.secType == "STK" and item.position > 0 + } + ib.disconnect() + return data + + return asyncio.run(_fetch()) + except Exception: + return None + + +def main(): + if not LEDGER.exists(): + print("还没有交易记录（trades.jsonl 不存在）") + return + + day, fills, realized_sym, realized_strat, lots, unknown_cost = load_and_replay() + + print(f"========== {day} 成交明细 ==========") + if not fills: + print(" 当天无成交") + for ts, strat, sym, action, qty, price, pnl in fills: + line = f"{ts} {short(strat):10s} {sym:6s} {action:4s} x{qty:g} @ {price:>10.2f}" + if pnl is not None: + line += f" 盈亏 {pnl:+.2f}" + elif action == "SELL": + line += " 盈亏 ? (成本未知)" + print(line) + + print(f"\n========== {day} 已实现盈亏 ==========") + if realized_sym: + print(" 按标的:") + for sym, v in sorted(realized_sym.items()): + print(f" {sym:6s} {v:+10.2f}") + if realized_strat: + print(" 按策略:") + for strat, v in sorted(realized_strat.items()): + print(f" {short(strat):10s} {v:+10.2f}") + total = sum(realized_sym.values()) + print(f" 合计: {total:+.2f}") + if unknown_cost: + print(f" (另有 {unknown_cost:g} 股卖出成本未知，未计入)") + + open_lots = [(s, sym, q, p) for (s, sym), dq in lots.items() for q, p in dq if q > 1e-9] + print("\n========== 当前持仓（成本价） ==========") + if not open_lots: + print(" 无") + else: + live = live_positions() + for strat, sym, qty, price in sorted(open_lots, key=lambda x: (x[1], x[0])): + line = f" {short(strat):10s} {sym:6s} x{qty:g} @ {price:>10.2f}" + if live and sym in live: + _, mkt, _ = live[sym] + line += f" 现价 {mkt:>9.2f} 浮动 {(mkt - price) * qty:+.2f}" + print(line) + if not live: + print(" (实时价格不可用)") + + +if __name__ == "__main__": + main() diff --git a/main.py b/main.py new file mode 100644 index 0000000..609ca87 --- /dev/null +++ b/main.py @@ -0,0 +1,220 @@ +import asyncio +import faulthandler +import logging +import signal +from datetime import datetime + +from bars import BarManager +from config import config +from connection import ib_conn +from state import PositionTracker +from strategies import MAStockStrategy, ForexMAStrategy, ShortTermMAVWAPStrategy, MeanReversionStrategy + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + datefmt="%H:%M:%S", +) +# silence ib_insync's per-message INFO spam (portfolio updates etc.) +logging.getLogger("ib_insync.wrapper").setLevel(logging.WARNING) +logging.getLogger("ib_insync.client").setLevel(logging.WARNING) +logger = logging.getLogger("main") + +# dump stack traces to stderr on SIGUSR1 (diagnostics for a hung main loop) +faulthandler.register(signal.SIGUSR1, all_threads=True) + + +class TradingApp: + def __init__(self): + self.strategies: list = [] + self._running = False + self._reconnect_task: asyncio.Task | None = None + self.tracker = PositionTracker(config.state_file) + self.bar_manager = BarManager(ib_conn.ib) + + def setup_signal_handlers(self): + for sig in (signal.SIGINT, signal.SIGTERM): + signal.signal(sig, self._signal_handler) + + def _signal_handler(self, sig, frame): + logger.info("Received signal %s, shutting down...", sig) + self._running = False + + async def connect(self) -> bool: + retries = 0 + while retries < config.max_retries: + if await ib_conn.ensure_connected(): + logger.info( + "Connected to IB Gateway at %s:%s (%s)", + config.ib.host, config.ib.port, config.ib.port_label, + ) + return True + retries += 1 + if retries < config.max_retries: + logger.warning("Retry %d/%d in %ds...", retries, config.max_retries, config.retry_delay) + await asyncio.sleep(config.retry_delay) + logger.error("Failed to connect after %d retries", config.max_retries) + return False + + def _account_position_keys(self) -> dict[str, float]: + """Real account positions keyed like the tracker: 'META' or 'EUR.USD'.""" + result: dict[str, float] = {} + for p in ib_conn.ib.positions(): + c = p.contract + key = f"{c.symbol}.{c.currency}" if c.secType == "CASH" else c.symbol + result[key] = result.get(key, 0) + p.position + return result + + async def _completed_stp_fills(self) -> dict[str, float]: + """{orderRef: avgFillPrice} for today's filled STP orders, so reconcile can + backfill the ledger with real prices for fills the bot missed while offline.""" + fills: dict[str, float] = {} + try: + for t in await ib_conn.ib.reqCompletedOrdersAsync(apiOnly=True): + ref = getattr(t.order, "orderRef", "") or "" + fill_price = getattr(t.orderStatus, "avgFillPrice", 0) or 0 + if ref.endswith(":STP") and t.orderStatus.status == "Filled" and fill_price > 0: + fills[ref] = fill_price + # ib_insync inserts completed orders into wrapper.trades with a + # non-terminal OrderStatus (filled=0, status from orderState). + # Purge them so they don't masquerade as open orders: has_open_order + # would block stop placement forever for a key whose real STP is + # missing but whose completed order is still visible (08-04: NOK + # zombie STP appeared "PreSubmitted" in openTrades after restart). + ib_conn.ib.wrapper.trades.pop(t.order.permId, None) + except Exception as e: + logger.warning("reqCompletedOrders failed: %s", e) + return fills + + async def run(self): + self.setup_signal_handlers() + self._running = True + + if config.ib.is_paper: + logger.info("Starting Trading Bot | Account: %s | Mode: PAPER", config.ib.account) + else: + logger.warning("*" * 70) + logger.warning( + "LIVE TRADING MODE - real orders will be placed! Account=%s Gateway=%s:%s (%s)", + config.ib.account, config.ib.host, config.ib.port, config.ib.port_label, + ) + logger.warning("*" * 70) + + if not await self.connect(): + logger.error("Initial connection failed, entering reconnect loop (Ctrl+C to abort)...") + await self._handle_disconnect() + if not self._running or not ib_conn.is_connected(): + return + + # registered exactly once; guarded against shutdown and duplicates + ib_conn.ib.disconnectedEvent += self._on_disconnected + + if config.stock.enabled: + self.strategies.append(MAStockStrategy(ib_conn.ib, self.bar_manager, self.tracker)) + if config.short_term.enabled: + self.strategies.append(ShortTermMAVWAPStrategy(ib_conn.ib, self.bar_manager, self.tracker)) + if config.forex.enabled: + self.strategies.append(ForexMAStrategy(ib_conn.ib, self.bar_manager, self.tracker)) + if config.mean_reversion.enabled: + self.strategies.append(MeanReversionStrategy(ib_conn.ib, self.bar_manager, self.tracker)) + + if not self.strategies: + logger.error("No strategies enabled") + return + + # reconcile tracked ownership against real account positions before trading + external_fills = await self._completed_stp_fills() + self.tracker.reconcile(self._account_position_keys(), external_fills) + + # populate open orders so strategies can adopt existing GTC stop orders + try: + await ib_conn.ib.reqOpenOrdersAsync() + except Exception as e: + logger.warning("reqOpenOrders failed: %s", e) + + for s in self.strategies: + await s.on_start() + logger.info("Strategy loaded: %s", s.name) + + logger.info("Bot is running. Press Ctrl+C to stop. Check interval: %.0fs", config.loop_interval) + + try: + while self._running: + # skip cycles while the reconnect task is working + if not ib_conn.is_connected(): + await asyncio.sleep(config.retry_delay) + continue + loop_start = datetime.now() + for s in self.strategies: + if not self._running: + break + try: + await s.on_bar() + except Exception as e: + logger.exception("Error in %s: %s", s.name, e) + + if not self._running: + break + + elapsed = (datetime.now() - loop_start).total_seconds() + await asyncio.sleep(max(0, config.loop_interval - elapsed)) + except asyncio.CancelledError: + pass + finally: + await self.shutdown() + + def _on_disconnected(self): + # ignore disconnects we triggered ourselves during shutdown + if not self._running: + return + # never run two reconnect loops concurrently + if self._reconnect_task and not self._reconnect_task.done(): + return + logger.warning("IB Gateway disconnected! Attempting reconnect...") + self._reconnect_task = asyncio.create_task(self._handle_disconnect()) + + async def _handle_disconnect(self): + attempt = 0 + while self._running: + attempt += 1 + delay = min(config.retry_delay * attempt, 60) + logger.warning("Reconnect attempt %d, waiting %ds...", attempt, delay) + await asyncio.sleep(delay) + if not self._running: + return + if await ib_conn.ensure_connected(): + logger.info("Reconnected successfully on attempt %d", attempt) + # live bar subscriptions died with the connection; re-subscribe + self.bar_manager.reset() + external_fills = await self._completed_stp_fills() + self.tracker.reconcile(self._account_position_keys(), external_fills) + try: + await ib_conn.ib.reqOpenOrdersAsync() + except Exception as e: + logger.warning("reqOpenOrders failed after reconnect: %s", e) + for s in self.strategies: + await s.on_start() + return + + async def shutdown(self): + logger.info("Shutting down...") + self._running = False + if self._reconnect_task and not self._reconnect_task.done(): + self._reconnect_task.cancel() + for s in self.strategies: + try: + await s.on_stop() + except Exception as e: + logger.exception("Error stopping %s: %s", s.name, e) + self.bar_manager.reset() + ib_conn.disconnect() + logger.info("Shutdown complete") + + +async def main(): + app = TradingApp() + await app.run() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/orders.py b/orders.py new file mode 100644 index 0000000..ce86fe8 --- /dev/null +++ b/orders.py @@ -0,0 +1,119 @@ +import asyncio +import logging +from typing import Optional + +from ib_insync import IB, Contract, MarketOrder, Trade + +from config import config + +logger = logging.getLogger(__name__) + + +def create_stock_contract(symbol: str, exchange: str = "SMART", currency: str = "USD") -> Contract: + return Contract(symbol=symbol, secType="STK", exchange=exchange, currency=currency) + + +def has_open_order(ib: IB, order_ref: str) -> bool: + """True if an unfinished order with this orderRef already exists.""" + for trade in ib.openTrades(): + if (trade.order.orderRef == order_ref and not trade.isDone() + and trade.orderStatus.remaining > 0): + return True + return False + + +async def wait_trade_done(trade: Trade, timeout: float) -> bool: + """Event-driven wait for a trade to reach a terminal state. + + Waits on trade.statusEvent (which fires on every status update) instead of + polling, returning as soon as isDone() becomes true or the timeout elapses. + """ + done = asyncio.Event() + + def _on_status(_): + if trade.isDone(): + done.set() + + trade.statusEvent.connect(_on_status) + try: + if trade.isDone(): + return True + try: + await asyncio.wait_for(done.wait(), timeout) + except asyncio.TimeoutError: + return False + return trade.isDone() + finally: + trade.statusEvent.disconnect(_on_status) + + +async def execute_market_order( + ib: IB, + contract: Contract, + action: str, + quantity: float, + order_ref: str, + timeout: Optional[float] = None, +) -> Optional[Trade]: + """Place a market order and wait for the fill. + + Returns the Trade if (partially) filled, None on rejection/timeout. + On timeout the order is cancelled before giving up. + """ + if has_open_order(ib, order_ref): + logger.warning( + "Skip %s %s x%g: an open order already exists (ref=%s)", + action, contract.symbol, quantity, order_ref, + ) + return None + + order = MarketOrder(action, quantity) + if config.ib.account: + order.account = config.ib.account + order.orderRef = order_ref + order.tif = "DAY" + order.outsideRth = False + + trade = ib.placeOrder(contract, order) + logger.info( + "Placed %s market order: %s x%g (ref=%s, orderId=%s)", + action, contract.symbol, quantity, order_ref, trade.order.orderId, + ) + + timeout = timeout or config.order_timeout + if not await wait_trade_done(trade, timeout): + logger.warning( + "%s %s x%g not filled within %.0fs, cancelling...", + action, contract.symbol, quantity, timeout, + ) + ib.cancelOrder(order) + await wait_trade_done(trade, 2.0) # wait for the cancel to confirm + + if trade.isDone(): + status = trade.orderStatus.status + if status == "Filled": + logger.info( + "%s %s x%g FILLED @ %.4f", + action, contract.symbol, trade.orderStatus.filled, trade.orderStatus.avgFillPrice, + ) + return trade + logger.error( + "%s %s x%g finished with status=%s: %s", + action, contract.symbol, quantity, status, + [m.message for m in trade.log] if trade.log else "", + ) + return None + + if trade.orderStatus.filled > 0: + logger.info( + "%s %s partially filled x%g @ %.4f before cancel", + action, contract.symbol, trade.orderStatus.filled, trade.orderStatus.avgFillPrice, + ) + return trade + + logger.error( + "%s %s x%g cancelled/unfilled (status=%s): %s", + action, contract.symbol, quantity, trade.orderStatus.status, + [m.message for m in trade.log] if trade.log else "", + ) + return None diff --git a/report.sh b/report.sh new file mode 100755 index 0000000..57db5cf --- /dev/null +++ b/report.sh @@ -0,0 +1,5 @@ +#!/bin/bash +# 查看当天（或指定日期）的成交明细与盈亏 +# 用法: ./report.sh [YYYY-MM-DD] +DIR="$(cd "$(dirname "$0")" && pwd)" +exec "$DIR/.venv/bin/python" "$DIR/daily_report.py" "$@" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..c5649c8 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +ib_insync>=0.9.86 +pandas>=2.0.0 +numpy>=1.24.0 +python-dotenv>=1.0.0 diff --git a/restart_bot.sh b/restart_bot.sh new file mode 100755 index 0000000..3c657ed --- /dev/null +++ b/restart_bot.sh @@ -0,0 +1,28 @@ +#!/bin/bash +set -e + +BOT_DIR="$(cd "$(dirname "$0")" && pwd)" +LOG_FILE="$BOT_DIR/trading_bot.log" + +echo "=== IB Trading Bot Starter ===" +echo "BOT_DIR: $BOT_DIR" + +# Step 1: Kill any existing bot process +pkill -f "main.py" 2>/dev/null || true +echo "[1/2] Stopped old bot process (if any)" + +# Step 2: Start the trading bot (setsid so it survives this shell exiting) +echo "[2/2] Starting trading bot..." +setsid nohup "$BOT_DIR/.venv/bin/python" "$BOT_DIR/main.py" >> "$LOG_FILE" 2>&1 < /dev/null & +BOT_PID=$! +echo "Bot PID: $BOT_PID" + +# Wait a few seconds and check +sleep 5 +if kill -0 "$BOT_PID" 2>/dev/null; then + echo "=== Bot started successfully (PID: $BOT_PID) ===" + echo "Log: tail -f $LOG_FILE" +else + echo "=== Bot failed to start, check log ===" + tail -5 "$LOG_FILE" +fi diff --git a/start_gateway.sh b/start_gateway.sh new file mode 100755 index 0000000..7c569ab --- /dev/null +++ b/start_gateway.sh @@ -0,0 +1,16 @@ +#!/bin/bash +# 启动 IB Gateway +# 确保先登录 IB 账户并开启 API 连接 + +JDK_LINK="$HOME/jdk" + +if [ ! -x "$JDK_LINK/bin/java" ]; then + echo "ERROR: JDK not found at $JDK_LINK" + echo "Download Temurin JDK and link it: ln -sfn ~/jdk- ~/jdk" + exit 1 +fi + +export JAVA_HOME="$JDK_LINK" +export PATH="$JAVA_HOME/bin:$PATH" + +~/ibgateway/ibgateway "$@" diff --git a/state.py b/state.py new file mode 100644 index 0000000..e3b2600 --- /dev/null +++ b/state.py @@ -0,0 +1,320 @@ +import json +import logging +import os +import tempfile +import time +from datetime import date, datetime, timezone +from pathlib import Path +from typing import Optional + +import config as app_config + +logger = logging.getLogger(__name__) + + +def _atomic_write_text(path: Path, text: str): + """Write text atomically: temp file in the same dir, then os.replace. + + Prevents corruption of the state files if the bot crashes (or the box + loses power) mid-write - os.replace is atomic on POSIX. + """ + fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp") + try: + with os.fdopen(fd, "w") as f: + f.write(text) + os.replace(tmp, path) + except Exception: + try: + os.unlink(tmp) + except OSError: + pass + raise + + +class PositionTracker: + """Tracks which strategy owns which positions, persisted to a JSON file. + + IB only reports account-level positions, so ownership is tracked locally: + {strategy_name: {symbol_or_pair: {"quantity": float, "entry_price": float, + "entry_date": "YYYY-MM-DD"}}} + """ + + def __init__(self, state_file: str): + self.state_file = Path(state_file) + self.ledger_file = self.state_file.with_name("trades.jsonl") + self._day_pnl_file = self.state_file.with_name("day_pnl.json") + self._recent_sells_file = self.state_file.with_name("recent_sells.json") + self._data: dict[str, dict[str, dict]] = {} + self.load() + self._seed_ledger_once() + self._day_date, self._day_realized = self._load_day_pnl() + self._recent_sells: dict[str, dict] = self._load_recent_sells() + + # ---------- daily realized P&L (for the max_daily_loss circuit breaker) ---------- + + def _load_day_pnl(self) -> tuple[str, float]: + try: + d = json.loads(self._day_pnl_file.read_text()) + if d.get("date") == date.today().isoformat(): + return d["date"], float(d.get("realized", 0.0)) + except Exception: + pass + return date.today().isoformat(), 0.0 + + def _save_day_pnl(self): + try: + _atomic_write_text( + self._day_pnl_file, + json.dumps({"date": self._day_date, "realized": self._day_realized}), + ) + except Exception as e: + logger.error("Failed to save day pnl: %s", e) + + def _add_day_pnl(self, pnl: float): + today = date.today().isoformat() + if self._day_date != today: + self._day_date = today + self._day_realized = 0.0 + self._day_realized += pnl + self._save_day_pnl() + + def day_realized_pnl(self) -> float: + """Today's realized P&L in USD (resets at local midnight).""" + if self._day_date != date.today().isoformat(): + return 0.0 + return self._day_realized + + # ---------- global sell registry (cross-strategy re-entry guard) ---------- + # + # Every sell of a symbol (any strategy, any reason, incl. backfilled + # external STP fills) is stamped here so no strategy can flip-flop: + # buy-sell-buy at the same price in a few minutes. + + def _load_recent_sells(self) -> dict[str, dict]: + try: + d = json.loads(self._recent_sells_file.read_text()) + if isinstance(d, dict): + return {k: v for k, v in d.items() + if isinstance(v, dict) and v.get("ts") is not None} + except Exception: + pass + return {} + + def _save_recent_sells(self): + try: + _atomic_write_text(self._recent_sells_file, json.dumps(self._recent_sells)) + except Exception as e: + logger.error("Failed to save recent sells: %s", e) + + def _record_recent_sell(self, key: str, price: float): + self._recent_sells[key] = { + "ts": datetime.now(timezone.utc).timestamp(), + "price": price, + } + self._save_recent_sells() + + def recent_sell(self, key: str) -> Optional[dict]: + """Latest sell record {ts (epoch), price} for the symbol, or None once + older than the improvement window (stale entries are purged lazily).""" + rec = self._recent_sells.get(key) + if not rec: + return None + window = app_config.config.sell_improvement_window_minutes * 60 + if time.time() - rec["ts"] > window: + del self._recent_sells[key] + self._save_recent_sells() + return None + return rec + + # ---------- trade ledger (append-only JSONL, used by daily_report.py) ---------- + + def _ledger_append(self, type_: str, strategy: str, key: str, qty: float, price: float, + estimated: bool = False): + rec = { + "ts": datetime.now().isoformat(timespec="seconds"), + "type": type_, "strategy": strategy, "symbol": key, + "qty": qty, "price": price, + } + if estimated: + rec["est"] = True + try: + with open(self.ledger_file, "a") as f: + f.write(json.dumps(rec) + "\n") + except Exception as e: + logger.error("Failed to append trade ledger: %s", e) + + def _seed_ledger_once(self): + """On first run after the ledger was introduced, record currently tracked + positions as opening lots so future sells can compute realized P&L.""" + if self.ledger_file.exists(): + return + for strategy, entries in self._data.items(): + for key, e in entries.items(): + self._ledger_append("seed", strategy, key, e["quantity"], e["entry_price"]) + if self._data: + logger.info("Trade ledger seeded with %d opening lots -> %s", + sum(len(v) for v in self._data.values()), self.ledger_file) + + def load(self): + if not self.state_file.exists(): + self._data = {} + return + try: + self._data = json.loads(self.state_file.read_text()) + logger.info("Loaded state from %s", self.state_file) + except Exception as e: + logger.error("Failed to load state file %s: %s (starting empty)", self.state_file, e) + self._data = {} + + def save(self): + try: + _atomic_write_text(self.state_file, json.dumps(self._data, indent=2)) + except Exception as e: + logger.error("Failed to save state file %s: %s", self.state_file, e) + + def get(self, strategy: str, key: str) -> Optional[dict]: + """Return owned position dict or None.""" + entry = self._data.get(strategy, {}).get(key) + if entry and entry.get("quantity", 0) > 0: + return entry + return None + + def total_positions(self) -> int: + """Number of open lots across all strategies (for the global position cap).""" + return sum( + 1 + for entries in self._data.values() + for e in entries.values() + if e.get("quantity", 0) > 0 + ) + + def symbol_value(self, key: str) -> float: + """Total tracked position value (qty x entry price) for one symbol, + summed across all strategies (for the per-symbol value cap).""" + return sum( + e["quantity"] * e["entry_price"] + for entries in self._data.values() + for k, e in entries.items() + if k == key and e.get("quantity", 0) > 0 + ) + + def record_buy(self, strategy: str, key: str, quantity: float, price: float): + entries = self._data.setdefault(strategy, {}) + entry = entries.get(key) + if entry: + prev_qty = entry["quantity"] + total_qty = prev_qty + quantity + entry["entry_price"] = (entry["entry_price"] * prev_qty + price * quantity) / total_qty + entry["quantity"] = total_qty + else: + entries[key] = { + "quantity": quantity, + "entry_price": price, + "entry_date": date.today().isoformat(), + "entry_ts": datetime.now(timezone.utc).isoformat(), + } + self.save() + self._ledger_append("buy", strategy, key, quantity, price) + logger.info("Tracker: %s owns %s x%g @ %.4f", strategy, key, entries[key]["quantity"], entries[key]["entry_price"]) + + def record_sell(self, strategy: str, key: str, quantity: float, price: float = None): + entries = self._data.get(strategy, {}) + entry = entries.get(key) + if not entry: + return + entry["quantity"] -= quantity + if entry["quantity"] <= 0: + del entries[key] + self.save() + if price is not None: + self._add_day_pnl((price - entry["entry_price"]) * quantity) + self._ledger_append("sell", strategy, key, quantity, price) + self._record_recent_sell(key, price) + logger.info("Tracker: %s sold %s x%g, remaining=%s", strategy, key, quantity, entry.get("quantity", 0)) + + def _record_external_sell(self, strategy: str, key: str, qty: float, + entry_price: float, external_fills: Optional[dict]): + """A tracked position vanished without a bot-recorded sell (e.g. its STP + order filled while the bot was disconnected). Backfill the ledger so the + report and day-PnL stay accurate: use the real fill price when IB can + provide it, otherwise estimate at the stop price and mark it.""" + price = (external_fills or {}).get(f"{strategy}:{key}:STP") + estimated = not price # None or 0.0 (IB may report 0.0 for cross-session fills) + if estimated: + price = round(entry_price * 0.98, 2) # rough STP fill estimate + logger.warning( + "Tracker: backfilling external sell %s %s x%g @ %.2f%s", + strategy, key, qty, price, " (ESTIMATED)" if estimated else "", + ) + self._ledger_append("sell_external", strategy, key, qty, price, estimated=estimated) + self._add_day_pnl((price - entry_price) * qty) + self._record_recent_sell(key, price) + + def reconcile(self, actual: dict[str, float], external_fills: Optional[dict] = None): + """Reconcile tracked state against real account positions. + + actual: {symbol_or_pair: total quantity held in account}. + external_fills: {orderRef: avgFillPrice} for STP fills retrieved from IB. + - tracked entries with no real position are cleared (ledger backfilled) + - tracked quantities exceeding the real position are clamped (ledger backfilled) + - real positions with no owner are reported as UNMANAGED (never sold by the bot) + """ + changed = False + for strategy in list(self._data): + entries = self._data[strategy] + for key in list(entries): + if actual.get(key, 0) <= 0: + entry = entries[key] + logger.warning( + "Tracker: %s owns %s x%g but account holds none - clearing stale state", + strategy, key, entry["quantity"], + ) + self._record_external_sell(strategy, key, entry["quantity"], + entry["entry_price"], external_fills) + del entries[key] + changed = True + + def tracked_totals(): + totals: dict[str, float] = {} + for entries in self._data.values(): + for key, e in entries.items(): + totals[key] = totals.get(key, 0) + e["quantity"] + return totals + + for key, total in tracked_totals().items(): + avail = actual.get(key, 0) + if total > avail: + logger.warning( + "Tracker: tracked %s x%g exceeds account position %g - clamping", + key, total, avail, + ) + remaining = avail + for strategy, entries in self._data.items(): + if key not in entries: + continue + entry = entries[key] + keep = min(entry["quantity"], remaining) + removed = entry["quantity"] - keep + if removed > 1e-9: + self._record_external_sell(strategy, key, removed, + entry["entry_price"], external_fills) + entry["quantity"] = keep + changed = True + remaining -= keep + if entry["quantity"] <= 0: + del entries[key] + changed = True + + tracked = tracked_totals() + for key, qty in actual.items(): + if qty > tracked.get(key, 0): + logger.warning( + "Tracker: account holds %s x%g but only %g tracked - " + "%g unit(s) UNMANAGED (bot will never sell them; " + "manage manually or add them to %s)", + key, qty, tracked.get(key, 0), qty - tracked.get(key, 0), self.state_file, + ) + + if changed: + self.save() + logger.info("State reconciled with account: %s", self._data) diff --git a/stop_bot_eod.sh b/stop_bot_eod.sh new file mode 100755 index 0000000..1f35012 --- /dev/null +++ b/stop_bot_eod.sh @@ -0,0 +1,16 @@ +#!/bin/bash +# One-shot task: gracefully stop the trading bot just before market close (15:59 ET), +# so no signals can fire after hours and queue orders for the next open. +# Self-removes from crontab after running. +BOT_DIR="$(cd "$(dirname "$0")" && pwd)" +LOG="$BOT_DIR/trading_bot.log" + +echo "=== $(date) EOD stop requested ===" >> "$LOG" +if pkill -TERM -f "python.*main\.py"; then + echo "$(date) SIGTERM sent to bot (graceful shutdown, may take up to 60s)" >> "$LOG" +else + echo "$(date) no bot process found" >> "$LOG" +fi + +# self-remove from crontab (one-shot task) +crontab -l 2>/dev/null | grep -v "stop_bot_eod" | crontab - diff --git a/strategies/__init__.py b/strategies/__init__.py new file mode 100644 index 0000000..008fdfa --- /dev/null +++ b/strategies/__init__.py @@ -0,0 +1,7 @@ +from .base import BaseStrategy +from .ma_cross import MAStockStrategy +from .forex import ForexMAStrategy +from .short_term import ShortTermMAVWAPStrategy +from .mean_reversion import MeanReversionStrategy + +__all__ = ["BaseStrategy", "MAStockStrategy", "ForexMAStrategy", "ShortTermMAVWAPStrategy", "MeanReversionStrategy"] diff --git a/strategies/base.py b/strategies/base.py new file mode 100644 index 0000000..9368545 --- /dev/null +++ b/strategies/base.py @@ -0,0 +1,351 @@ +import logging +import time +from abc import ABC, abstractmethod +from datetime import date, datetime, time as dtime, timedelta +from typing import Optional + +import pandas as pd +from ib_insync import IB, Contract, StopOrder, Trade + +from bars import BarManager +from config import config +from orders import has_open_order, wait_trade_done +from state import PositionTracker + +logger = logging.getLogger(__name__) + + +class BaseStrategy(ABC): + def __init__(self, ib: IB, bar_manager: BarManager, tracker: PositionTracker): + self.ib = ib + self.bar_manager = bar_manager + self.tracker = tracker + self.name = self.__class__.__name__ + self._stop_trades: dict[str, Trade] = {} + self._stop_consumed: set = set() # uids of done stop orders already processed + self._stop_cooldown: dict[str, datetime] = {} # key -> last stop-out time + self._peak_high: dict[str, float] = {} # key -> highest high since entry (trailing) + + @abstractmethod + async def on_tick(self): + pass + + @abstractmethod + async def on_bar(self): + pass + + async def on_start(self): + pass + + async def on_stop(self): + pass + + def __repr__(self) -> str: + return f"<{self.name}>" + + def _order_quantity(self, key: str, price: float) -> int: + """Share count targeting the (per-symbol) dollar value per trade (min 1).""" + value = self._trade_value_usd(key) + if price and price > 0: + return max(1, int(value // price)) + return 1 + + def _trade_value_usd(self, key: str) -> float: + """Per-trade dollar value for a symbol; global per-symbol override wins.""" + override = config.symbol_trade_value_usd.get(key) + return override if override else self.cfg.trade_value_usd + + def _can_open_position(self, key: str, order_value: float, price: float = 0.0) -> tuple[bool, str]: + """Global entry gates shared by all strategies. + + Returns (allowed, reason_if_blocked): + - sell_only mode -> no new positions at all + - daily realized loss hit max_daily_loss -> no new positions today + - concurrent lots hit max_positions -> no new positions until some close + - symbol's combined position value would exceed max_symbol_value_usd + - symbol sold by ANY strategy within sell_cooldown_minutes -> no re-entry + - beyond the cooldown (within sell_improvement_window_minutes), a re-buy + is only allowed if its price is >= sell_improvement_pct below the + last sell price (prevents same-price flip-flops across strategies) + """ + rec = self.tracker.recent_sell(key) + if rec is not None: + age_min = (time.time() - rec["ts"]) / 60 + if age_min < config.sell_cooldown_minutes: + return False, (f"{key} sold {age_min:.0f} min ago by another strategy - " + f"global sell cooldown {config.sell_cooldown_minutes} min") + if price > 0 and age_min < config.sell_improvement_window_minutes: + need = rec["price"] * (1 - config.sell_improvement_pct / 100) + if price >= need: + return False, (f"{key} re-buy price {price:.2f} not >= " + f"{config.sell_improvement_pct:.1f}% below last sell " + f"{rec['price']:.2f} (need < {need:.2f})") + if config.sell_only: + return False, "sell-only mode (liquidating)" + day_pnl = self.tracker.day_realized_pnl() + if day_pnl <= -config.max_daily_loss: + return False, f"daily loss limit hit (realized {day_pnl:+.2f} <= -{config.max_daily_loss:.0f})" + if self.tracker.total_positions() >= config.max_positions: + return False, f"global cap of {config.max_positions} positions reached" + symbol_value = self.tracker.symbol_value(key) + if symbol_value + order_value > config.max_symbol_value_usd: + return False, (f"{key} value cap: ${symbol_value + order_value:,.0f} would exceed " + f"${config.max_symbol_value_usd:,.0f}") + return True, "" + + # ---------- stop-out / order-failure cooldown (per strategy+symbol) ---------- + + @staticmethod + def _next_day_start() -> datetime: + """End-of-day boundary: next local midnight (rest of today = no re-entry).""" + return datetime.combine(date.today() + timedelta(days=1), dtime.min) + def _mark_cooldown(self, key: str, reason: str, until: Optional[datetime] = None): + expiry = until or (datetime.now() + timedelta(minutes=self.cfg.stop_cooldown_minutes)) + self._stop_cooldown[key] = expiry + desc = "until EOD" if until else f"{self.cfg.stop_cooldown_minutes} min" + logger.info("%s: %s cooldown for %s (%s, %s)", self.name, key, key, desc, reason) + + def _mark_stop_cooldown(self, key: str): + self._mark_cooldown(key, "stop-out") + + def _mark_hard_stop_cooldown(self, key: str): + """After a hard-stop fill, block re-entry on this symbol for the WHOLE day + (prevents same-price re-buys after an overnight-gap stop-out).""" + self._mark_cooldown(key, "hard stop-out", until=self._next_day_start()) + + def _in_stop_cooldown(self, key: str) -> bool: + until = self._stop_cooldown.get(key) + if until is None: + return False + if datetime.now() >= until: + del self._stop_cooldown[key] + return False + return True + + # ---------- exchange-side hard stop (GTC STP order) management ---------- + + def _use_hard_stop(self) -> bool: + return getattr(self.cfg, "use_hard_stop", False) + + def _stop_ref(self, key: str) -> str: + return f"{self.name}:{key}:STP" + + def _has_active_stop(self, key: str) -> bool: + t = self._stop_trades.get(key) + return t is not None and not t.isDone() + + # ---------- trailing stop: raise the stop as price makes new highs ---------- + + def _use_trailing_stop(self) -> bool: + return bool(getattr(self.cfg, "use_trailing_stop", False)) + + def _target_stop_price(self, key: str, entry_price: float, peak_high: float | None) -> float: + """Desired STP price for a position. + + Fixed stop = entry - stop_loss_pct (always the floor). + Once the peak high since entry is >= min_profit above entry, switch to a + trailing stop = peak_high - trailing_stop_pct, but never below the fixed stop. + """ + fixed = round(entry_price * (1 - self.cfg.stop_loss_pct / 100), 2) + if not self._use_trailing_stop() or not peak_high: + return fixed + peak_profit_pct = (peak_high - entry_price) / entry_price * 100 + if peak_profit_pct < self.cfg.min_profit_pct: + return fixed + trailing = round(peak_high * (1 - self.cfg.trailing_stop_pct / 100), 2) + return max(trailing, fixed) + + def _track_peak_high(self, key: str, high: float): + """Remember the highest high seen for a symbol (call with each completed bar).""" + current = self._peak_high.get(key) + if current is None or high > current: + self._peak_high[key] = high + + def _current_peak_high(self, key: str) -> float | None: + return self._peak_high.get(key) + + @staticmethod + def _trade_uid(trade: Trade): + """Stable unique id for a trade (permId once assigned by IB).""" + return trade.order.permId or trade.order.orderId or id(trade) + + async def _sync_stop_orders(self): + """Reconcile in-memory/exchange stop orders with tracked positions. + + Runs at the top of every on_bar cycle: + - STP filled -> record the sell in the tracker (position is gone) + - STP dead (cancelled/inactive) while position owned -> re-place next cycle + - STP open -> adopt it (e.g. after a bot restart); raise it to the + trailing target if it has moved up + - owned but no STP anywhere -> place a new one + """ + if not self._use_hard_stop(): + return + by_ref = {} + for t in self.ib.trades(): + ref = getattr(t.order, "orderRef", "") or "" + if ref: + by_ref[ref] = t + + for key, contract in self.contracts.items(): + owned = self.tracker.get(self.name, key) + trade = self._stop_trades.get(key) + if trade is None: + candidate = by_ref.get(self._stop_ref(key)) + if candidate is not None: + if not candidate.isDone(): + trade = candidate # adopt open order (e.g. after restart) + elif self._trade_uid(candidate) not in self._stop_consumed: + trade = candidate # consume terminal state exactly once + + if trade is not None and trade.isDone(): + self._stop_consumed.add(self._trade_uid(trade)) + status = trade.orderStatus.status + filled = trade.orderStatus.filled or 0.0 + if status == "Filled" and filled > 0: + if owned: + logger.info( + "%s HARD STOP filled: %s x%g @ %.2f", + self.name, key, filled, trade.orderStatus.avgFillPrice, + ) + self.tracker.record_sell(self.name, key, filled, trade.orderStatus.avgFillPrice) + self._mark_hard_stop_cooldown(key) + elif owned: + logger.warning( + "%s stop order for %s ended with status=%s - re-placing now", + self.name, key, status, + ) + self._stop_trades.pop(key, None) + if owned and not (status == "Filled" and filled > 0): + # re-place immediately (same cycle), don't wait for the next one + target = await self._current_stop_target(key, contract, owned) + await self._place_stop(key, contract, owned["quantity"], target) + continue + + if trade is not None: + self._stop_trades[key] = trade # adopt existing open order + if owned: + await self._raise_stop_if_needed(key, contract, owned) + continue + + if owned: + target = await self._current_stop_target(key, contract, owned) + await self._place_stop(key, contract, owned["quantity"], target) + + async def _current_stop_target(self, key: str, contract: Contract, owned: dict) -> float: + """Desired STP price for the position right now (fixed or trailing).""" + entry = owned["entry_price"] + peak = await self._peak_high_since_entry(key, contract, owned) + return self._target_stop_price(key, entry, peak) + + async def _peak_high_since_entry(self, key: str, contract: Contract, owned: dict) -> float | None: + """Highest high of completed bars since the position was opened, if any.""" + try: + df = await self._get_df(contract) + except Exception: + return None + if df is None or df.empty: + return None + cutoff = None + entry_ts = owned.get("entry_ts") + if entry_ts: + try: + cutoff = pd.Timestamp(entry_ts, tz="UTC") + except Exception: + cutoff = None + if cutoff is None: + ed = owned.get("entry_date") + if ed: + try: + cutoff = pd.Timestamp(ed, tz="America/New_York").tz_convert("UTC") + except Exception: + cutoff = None + if cutoff is None: + return None + since = df.loc[df["date"] >= cutoff, "high"] + return float(since.max()) if not since.empty else None + + async def _raise_stop_if_needed(self, key: str, contract: Contract, owned: dict): + """Cancel and re-place the stop if its price no longer matches the target. + + Handles both upward trailing moves and stale/zombie orders whose auxPrice + differs from the desired stop (e.g. a PreSubmitted order that IB never + accepted; cancel+re-place recovers it instead of silently keeping it). + """ + trade = self._stop_trades.get(key) + if trade is None or trade.isDone(): + return + target = await self._current_stop_target(key, contract, owned) + current = getattr(trade.order, "auxPrice", 0) or 0 + if current and abs(target - current) < 0.01: + return + logger.info( + "%s raising stop %s %.2f -> %.2f (trailing)", self.name, key, current, target, + ) + stop_filled, stop_price, confirmed = await self._cancel_stop(key) + if stop_filled > 0: + logger.info("%s: stop filled %g during trailing raise", key, stop_filled) + self.tracker.record_sell(self.name, key, stop_filled, stop_price) + remaining = owned["quantity"] - stop_filled + if remaining <= 0: + return + if not confirmed: + logger.warning( + "%s: trailing raise aborted for %s - stop cancel not confirmed", + self.name, key, + ) + return + await self._place_stop(key, contract, remaining, target) + + async def _place_stop(self, key: str, contract: Contract, quantity: float, stop_price: float): + """Place a GTC stop-loss sell order at the exchange.""" + stop_price = round(stop_price, 2) + ref = self._stop_ref(key) + if has_open_order(self.ib, ref): + _matching = [ + f"{t.order.orderRef}|{t.orderStatus.status}|client{t.order.clientId}|done={t.isDone()}" + for t in self.ib.openTrades() if t.order.orderRef == ref + ] + logger.info( + "%s skipping stop place for %s: matching open order exists %s (debug: owned=%s stop_trades=%s)", + self.name, key, _matching, + self.tracker.get(self.name, key), + self._stop_trades.get(key), + ) + return + order = StopOrder("SELL", quantity, stop_price) + if config.ib.account: + order.account = config.ib.account + order.orderRef = ref + order.tif = "GTC" + order.outsideRth = True # allow triggering in extended hours / overnight gaps + try: + trade = self.ib.placeOrder(contract, order) + self._stop_trades[key] = trade + logger.info( + "%s hard stop placed: %s x%g STP @ %.2f (GTC, outsideRth)", + self.name, key, quantity, stop_price, + ) + except Exception as e: + logger.error("%s failed to place stop for %s: %s", self.name, key, e) + async def _cancel_stop(self, key: str) -> tuple[float, float, bool]: + """Cancel the stop order for key. + + Returns (quantity, avg_price) filled in the cancel race, and whether the + cancellation was confirmed before giving up. If not confirmed, the caller + must NOT proceed with a market sell - the STP may still be live and both + orders could fill (double sell). Leave the position for next cycle. + """ + trade = self._stop_trades.pop(key, None) + if trade is None: + return 0.0, 0.0, True + if not trade.isDone(): + self.ib.cancelOrder(trade.order) + if not await wait_trade_done(trade, 5.0): + logger.warning( + "%s: cancel of stop for %s not confirmed after 5s - " + "assuming it may still be active", + self.name, key, + ) + return trade.orderStatus.filled or 0.0, trade.orderStatus.avgFillPrice or 0.0, False + return trade.orderStatus.filled or 0.0, trade.orderStatus.avgFillPrice or 0.0, True diff --git a/strategies/forex.py b/strategies/forex.py new file mode 100644 index 0000000..cf0b584 --- /dev/null +++ b/strategies/forex.py @@ -0,0 +1,120 @@ +import logging + +import pandas as pd +from ib_insync import Contract, IB + +from bars import BarManager, is_market_active, parse_bar_size_seconds, to_completed_df +from config import config +from orders import execute_market_order +from state import PositionTracker +from strategies.base import BaseStrategy + +logger = logging.getLogger(__name__) + + +class ForexMAStrategy(BaseStrategy): + """Forex MA cross strategy (disabled by default). + + Exits: hard stop loss at -stop_loss_pct, or fast MA crossing below slow MA. + """ + + def __init__(self, ib: IB, bar_manager: BarManager, tracker: PositionTracker): + super().__init__(ib, bar_manager, tracker) + self.cfg = config.forex + self.fast_period = self.cfg.fast_ma_period + self.slow_period = self.cfg.slow_ma_period + self.units = self.cfg.trade_units + self.bar_seconds = parse_bar_size_seconds(self.cfg.bar_size) + self.contracts: dict[str, Contract] = {} + + async def on_start(self): + logger.info( + "Forex MA strategy started: pairs=%s, fast=%d, slow=%d, units=%d, stop_loss=%.1f%%", + self.cfg.pairs, self.fast_period, self.slow_period, self.units, self.cfg.stop_loss_pct, + ) + for pair in self.cfg.pairs: + base, quote = pair.split(".") + contract = Contract(secType="CASH", symbol=base, currency=quote, exchange=self.cfg.exchange) + await self.ib.qualifyContractsAsync(contract) + if not contract.conId: + logger.error("Forex %s: failed to qualify contract, skipped", pair) + continue + self.contracts[pair] = contract + await self.bar_manager.subscribe( + contract, self.cfg.bar_size, f"{self.cfg.lookback_days} D", "MIDPOINT" + ) + logger.info("Forex %s: subscribed, owned=%s", pair, self.tracker.get(self.name, pair)) + + async def _get_df(self, contract: Contract) -> pd.DataFrame: + bars = self.bar_manager.get(contract, self.cfg.bar_size, "MIDPOINT") + if bars is None: + bars = await self.bar_manager.subscribe( + contract, self.cfg.bar_size, f"{self.cfg.lookback_days} D", "MIDPOINT" + ) + return to_completed_df(bars, self.bar_seconds) + + async def on_bar(self): + for pair, contract in self.contracts.items(): + try: + await self._process_pair(pair, contract) + except Exception as e: + logger.exception("Forex %s: error: %s", pair, e) + + async def _process_pair(self, pair: str, contract: Contract): + df = await self._get_df(contract) + if len(df) < self.slow_period + 2: + return + if not is_market_active(df, self.bar_seconds): + return + + df = df.copy() + df["fast_ma"] = df["close"].rolling(self.fast_period).mean() + df["slow_ma"] = df["close"].rolling(self.slow_period).mean() + + last = df.iloc[-1] + prev = df.iloc[-2] + fast_above = last["fast_ma"] > last["slow_ma"] + prev_fast_above = prev["fast_ma"] > prev["slow_ma"] + cross_up = fast_above and not prev_fast_above + cross_down = not fast_above and prev_fast_above + + owned = self.tracker.get(self.name, pair) + + if owned: + entry = owned["entry_price"] + if last["close"] <= entry * (1 - self.cfg.stop_loss_pct / 100): + logger.info( + "FOREX STOP-LOSS SELL: %s close=%.5f entry=%.5f", + pair, last["close"], entry, + ) + await self._sell(pair, contract, owned["quantity"]) + elif cross_down: + logger.info( + "FOREX SELL %s (fast MA %.5f < slow MA %.5f)", + pair, last["fast_ma"], last["slow_ma"], + ) + await self._sell(pair, contract, owned["quantity"]) + else: + if cross_up: + logger.info( + "FOREX BUY %s (fast MA %.5f crossed above slow MA %.5f)", + pair, last["fast_ma"], last["slow_ma"], + ) + await self._buy(pair, contract) + + async def _buy(self, pair: str, contract: Contract): + ref = f"{self.name}:{pair}" + trade = await execute_market_order(self.ib, contract, "BUY", self.units, ref) + if trade and trade.orderStatus.filled > 0: + self.tracker.record_buy( + self.name, pair, trade.orderStatus.filled, trade.orderStatus.avgFillPrice + ) + + async def _sell(self, pair: str, contract: Contract, quantity: float): + ref = f"{self.name}:{pair}" + trade = await execute_market_order(self.ib, contract, "SELL", quantity, ref) + if trade and trade.orderStatus.filled > 0: + self.tracker.record_sell(self.name, pair, trade.orderStatus.filled) + + async def on_tick(self): + pass diff --git a/strategies/ma_cross.py b/strategies/ma_cross.py new file mode 100644 index 0000000..6dfcd37 --- /dev/null +++ b/strategies/ma_cross.py @@ -0,0 +1,212 @@ +import logging + +import pandas as pd +from ib_insync import Contract, IB + +from bars import BarManager, is_market_active, parse_bar_size_seconds, to_completed_df +from config import config +from orders import execute_market_order +from state import PositionTracker +from strategies.base import BaseStrategy + +logger = logging.getLogger(__name__) + + +class MAStockStrategy(BaseStrategy): + """SMA fast/slow golden-cross strategy with ADX trend filter. + + Exits (re-evaluated every cycle while in position): + - hard stop loss at -stop_loss_pct (always honoured) + - fast MA below slow MA, once profit >= min_profit_pct + """ + + def __init__(self, ib: IB, bar_manager: BarManager, tracker: PositionTracker): + super().__init__(ib, bar_manager, tracker) + self.cfg = config.stock + self.fast_period = self.cfg.fast_ma_period + self.slow_period = self.cfg.slow_ma_period + self.bar_seconds = parse_bar_size_seconds(self.cfg.bar_size) + self.contracts: dict[str, Contract] = {} + + async def on_start(self): + logger.info( + "Stock MA strategy started: symbols=%s, fast=%d, slow=%d, value=$%.0f, " + "stop_loss=%.1f%%, min_profit=%.1f%%, confirm_bars=%d", + self.cfg.symbols, self.fast_period, self.slow_period, self.cfg.trade_value_usd, + self.cfg.stop_loss_pct, self.cfg.min_profit_pct, self.cfg.entry_confirm_bars, + ) + for symbol in self.cfg.symbols: + contract = self._contract(symbol) + await self.ib.qualifyContractsAsync(contract) + if not contract.conId: + logger.error("Stock %s: failed to qualify contract, skipped", symbol) + continue + self.contracts[symbol] = contract + await self.bar_manager.subscribe( + contract, self.cfg.bar_size, f"{self.cfg.lookback_days} D", "TRADES" + ) + logger.info("Stock %s: subscribed, owned=%s", symbol, self.tracker.get(self.name, symbol)) + + @staticmethod + def _calc_adx(df: pd.DataFrame, period: int = 14) -> pd.Series: + """ADX with standard Wilder smoothing.""" + high, low, close = df["high"], df["low"], df["close"] + prev_close = close.shift(1) + tr = pd.concat([ + (high - low).abs(), + (high - prev_close).abs(), + (low - prev_close).abs(), + ], axis=1).max(axis=1) + up_move = high.diff() + down_move = -low.diff() # Wilder: positive only when the low moves DOWN + plus_dm = ((up_move > down_move) & (up_move > 0)).astype(float) * up_move + minus_dm = ((down_move > up_move) & (down_move > 0)).astype(float) * down_move + atr = tr.ewm(alpha=1 / period, adjust=False).mean() + plus_di = 100 * plus_dm.ewm(alpha=1 / period, adjust=False).mean() / atr + minus_di = 100 * minus_dm.ewm(alpha=1 / period, adjust=False).mean() / atr + dx = 100 * (plus_di - minus_di).abs() / (plus_di + minus_di).replace(0, float("nan")) + return dx.ewm(alpha=1 / period, adjust=False).mean() + + async def _get_df(self, contract: Contract) -> pd.DataFrame: + bars = self.bar_manager.get(contract, self.cfg.bar_size, "TRADES") + if bars is None: + bars = await self.bar_manager.subscribe( + contract, self.cfg.bar_size, f"{self.cfg.lookback_days} D", "TRADES" + ) + return to_completed_df(bars, self.bar_seconds) + + async def on_bar(self): + await self._sync_stop_orders() + for symbol, contract in self.contracts.items(): + try: + await self._process_symbol(symbol, contract) + except Exception as e: + logger.exception("Stock %s: error: %s", symbol, e) + + async def _process_symbol(self, symbol: str, contract: Contract): + df = await self._get_df(contract) + if len(df) < self.slow_period + 28: + return + if not is_market_active(df, self.bar_seconds): + return # market closed / stale data - never trade + + df = df.copy() + df["fast_ma"] = df["close"].rolling(self.fast_period).mean() + df["slow_ma"] = df["close"].rolling(self.slow_period).mean() + df["adx"] = self._calc_adx(df) + + last = df.iloc[-1] + prev = df.iloc[-2] + fast_above = last["fast_ma"] > last["slow_ma"] + + # entry confirmation: the cross happened (confirm_bars-1) bars back and + # the fast MA is still above the slow MA now -> filters 1-bar whipsaws + cb = self.cfg.entry_confirm_bars + ref = df.iloc[-cb] + ref_prev = df.iloc[-cb - 1] + cross_then = ref["fast_ma"] > ref["slow_ma"] and ref_prev["fast_ma"] <= ref_prev["slow_ma"] + confirmed = cross_then and fast_above + + is_trending = last["adx"] > self.cfg.adx_min + slow_rising = True + if self.cfg.require_slow_ma_slope and len(df) >= 2: + # slow MA must itself be rising over the confirmation window + slow_rising = last["slow_ma"] > df["slow_ma"].iloc[-self.cfg.entry_confirm_bars - 1] + + owned = self.tracker.get(self.name, symbol) + + if owned: + entry = owned["entry_price"] + profit_pct = (last["close"] - entry) / entry * 100 + + # soft stop is only a fallback: the exchange-side STP order is primary + if not self._has_active_stop(symbol) and last["close"] <= entry * (1 - self.cfg.stop_loss_pct / 100): + logger.info( + "STOCK STOP-LOSS SELL (soft fallback): %s close=%.2f entry=%.2f (%.2f%%)", + symbol, last["close"], entry, profit_pct, + ) + await self._sell(symbol, contract, owned["quantity"]) + self._mark_stop_cooldown(symbol) + elif not fast_above: + # re-checked every cycle: exits as soon as profit requirement is met + if profit_pct >= self.cfg.min_profit_pct: + logger.info( + "STOCK SELL: %s (fast MA %.2f < slow MA %.2f, profit=%.2f%%, ADX=%.1f)", + symbol, last["fast_ma"], last["slow_ma"], profit_pct, last["adx"], + ) + await self._sell(symbol, contract, owned["quantity"]) + else: + logger.debug( + "STOCK SELL WAITING: %s profit %.2f%% < min %.1f%%", + symbol, profit_pct, self.cfg.min_profit_pct, + ) + else: + if confirmed and is_trending and slow_rising: + if self._in_stop_cooldown(symbol): + logger.info("STOCK BUY SKIPPED: %s (stop-out cooldown %d min)", + symbol, self.cfg.stop_cooldown_minutes) + return + value = self._trade_value_usd(symbol) + ok, reason = self._can_open_position(symbol, value, last["close"]) + if not ok: + logger.info("STOCK BUY SKIPPED: %s (%s)", symbol, reason) + return + qty = self._order_quantity(symbol, last["close"]) + logger.info( + "STOCK BUY: %s x%d (cross confirmed over %d bars, fast MA %.2f > slow MA %.2f, ADX=%.1f)", + symbol, qty, self.cfg.entry_confirm_bars, last["fast_ma"], last["slow_ma"], last["adx"], + ) + await self._buy(symbol, contract, qty) + + async def _buy(self, symbol: str, contract: Contract, quantity: int): + ref = f"{self.name}:{symbol}" + trade = await execute_market_order(self.ib, contract, "BUY", quantity, ref) + if trade and trade.orderStatus.filled > 0: + self.tracker.record_buy( + self.name, symbol, trade.orderStatus.filled, trade.orderStatus.avgFillPrice + ) + if self._use_hard_stop(): + stop_price = self._target_stop_price( + symbol, trade.orderStatus.avgFillPrice, None + ) + await self._place_stop( + symbol, contract, trade.orderStatus.filled, stop_price + ) + else: + # order rejected/timed out (e.g. insufficient buying power): + # cool down to avoid retrying every cycle + self._mark_cooldown(symbol, "order failed") + + async def _sell(self, symbol: str, contract: Contract, quantity: float): + # cancel the hard stop first; it may have filled in the cancel race + if self._use_hard_stop(): + stop_filled, stop_price, cancel_confirmed = await self._cancel_stop(symbol) + if stop_filled > 0: + logger.info("%s: stop order filled %g during cancel", symbol, stop_filled) + self.tracker.record_sell(self.name, symbol, stop_filled, stop_price) + quantity -= stop_filled + if quantity <= 0: + return + if not cancel_confirmed: + # STP may still be live - a market sell could double-sell; + # leave the position for the next cycle + logger.warning( + "STOCK SELL ABORTED: %s (stop cancel not confirmed - " + "may still be active, will retry next cycle)", symbol, + ) + return + ref = f"{self.name}:{symbol}" + trade = await execute_market_order(self.ib, contract, "SELL", quantity, ref) + if trade and trade.orderStatus.filled > 0: + self.tracker.record_sell(self.name, symbol, trade.orderStatus.filled, trade.orderStatus.avgFillPrice) + + async def on_tick(self): + pass + + def _contract(self, symbol: str): + return Contract( + symbol=symbol, + secType="STK", + exchange=self.cfg.exchange, + currency=self.cfg.currency, + ) diff --git a/strategies/mean_reversion.py b/strategies/mean_reversion.py new file mode 100644 index 0000000..f22d4cf --- /dev/null +++ b/strategies/mean_reversion.py @@ -0,0 +1,237 @@ +import logging + +import pandas as pd +from ib_insync import Contract, IB + +from bars import BarManager, is_market_active, parse_bar_size_seconds, to_completed_df +from config import config +from orders import execute_market_order +from state import PositionTracker +from strategies.base import BaseStrategy + +logger = logging.getLogger(__name__) + + +class MeanReversionStrategy(BaseStrategy): + """Buy oversold dips (RSI / lower Bollinger / rapid drop), sell on recovery. + + Exits (re-evaluated every cycle while in position): + - hard stop loss at -stop_loss_pct (always honoured, overrides min profit) + - RSI overbought / price back at Bollinger mid / recovery_pct reached, + once profit >= min_profit_pct + """ + + def __init__(self, ib: IB, bar_manager: BarManager, tracker: PositionTracker): + super().__init__(ib, bar_manager, tracker) + self.cfg = config.mean_reversion + self.bar_seconds = parse_bar_size_seconds(self.cfg.bar_size) + self.contracts: dict[str, Contract] = {} + + async def on_start(self): + logger.info( + "MeanReversion strategy started: symbols=%s, value=$%.0f, " + "RSI(%d, %.0f/%.0f), BB(%d, %.1f), drop=%.1f%%/%dbars, recovery=%.1f%%, " + "stop_loss=%.1f%%, confirm_bars=%d", + self.cfg.symbols, self.cfg.trade_value_usd, + self.cfg.rsi_period, self.cfg.rsi_oversold, self.cfg.rsi_overbought, + self.cfg.bb_period, self.cfg.bb_std, + self.cfg.rapid_drop_pct, self.cfg.rapid_drop_bars, self.cfg.recovery_pct, + self.cfg.stop_loss_pct, self.cfg.entry_confirm_bars, + ) + for symbol in self.cfg.symbols: + contract = self._contract(symbol) + await self.ib.qualifyContractsAsync(contract) + if not contract.conId: + logger.error("MeanRev %s: failed to qualify contract, skipped", symbol) + continue + self.contracts[symbol] = contract + await self.bar_manager.subscribe( + contract, self.cfg.bar_size, f"{self.cfg.lookback_days} D", "TRADES" + ) + logger.info("MeanRev %s: subscribed, owned=%s", symbol, self.tracker.get(self.name, symbol)) + + @staticmethod + def _calc_rsi(series: pd.Series, period: int = 14) -> pd.Series: + """RSI with standard Wilder smoothing.""" + delta = series.diff() + gain = delta.where(delta > 0, 0.0) + loss = -delta.where(delta < 0, 0.0) + avg_gain = gain.ewm(alpha=1 / period, adjust=False).mean() + avg_loss = loss.ewm(alpha=1 / period, adjust=False).mean() + # no-loss division yields inf -> RSI 100; 0/0 (flat) yields NaN -> no signal + rs = avg_gain / avg_loss + return 100 - (100 / (1 + rs)) + + @staticmethod + def _calc_bollinger(series: pd.Series, period: int = 20, num_std: float = 2.0): + mid = series.rolling(period).mean() + std = series.rolling(period).std() + upper = mid + num_std * std + lower = mid - num_std * std + return mid, upper, lower + + async def _get_df(self, contract: Contract) -> pd.DataFrame: + bars = self.bar_manager.get(contract, self.cfg.bar_size, "TRADES") + if bars is None: + bars = await self.bar_manager.subscribe( + contract, self.cfg.bar_size, f"{self.cfg.lookback_days} D", "TRADES" + ) + return to_completed_df(bars, self.bar_seconds) + + async def on_bar(self): + await self._sync_stop_orders() + for symbol, contract in self.contracts.items(): + try: + await self._process_symbol(symbol, contract) + except Exception as e: + logger.exception("MeanRev %s: error: %s", symbol, e) + + async def _process_symbol(self, symbol: str, contract: Contract): + df = await self._get_df(contract) + min_bars = max(self.cfg.rsi_period, self.cfg.bb_period, self.cfg.rapid_drop_bars, + self.cfg.trend_ma_period) + 5 + if len(df) < min_bars: + return + if not is_market_active(df, self.bar_seconds): + return # market closed / stale data - never trade + + df = df.copy() + df["rsi"] = self._calc_rsi(df["close"], self.cfg.rsi_period) + df["bb_mid"], df["bb_upper"], df["bb_lower"] = self._calc_bollinger( + df["close"], self.cfg.bb_period, self.cfg.bb_std + ) + df["trend_ma"] = df["close"].rolling(self.cfg.trend_ma_period).mean() + + last = df.iloc[-1] + owned = self.tracker.get(self.name, symbol) + + if not owned: + buy_signal = None + + # entry confirmation: the oversold condition must hold on the last + # confirm_bars consecutive bars -> filters 1-bar spikes + cb = self.cfg.entry_confirm_bars + recent_cb = df.iloc[-cb:] + + if (recent_cb["rsi"] < self.cfg.rsi_oversold).all(): + buy_signal = "RSI" + elif (recent_cb["close"] <= recent_cb["bb_lower"]).all(): + buy_signal = "Bollinger" + elif len(df) > self.cfg.rapid_drop_bars: + recent = df.iloc[-self.cfg.rapid_drop_bars - 1:] + price_change_pct = (recent.iloc[-1]["close"] - recent.iloc[0]["close"]) / recent.iloc[0]["close"] * 100 + if price_change_pct <= -self.cfg.rapid_drop_pct: + buy_signal = "RapidDrop" + + if buy_signal and last["close"] <= last["trend_ma"]: + # trend filter: don't catch falling knives below the slow SMA + logger.debug( + "MeanRev BUY blocked by trend filter: %s [%s] close %.2f <= SMA%d %.2f", + symbol, buy_signal, last["close"], self.cfg.trend_ma_period, last["trend_ma"], + ) + buy_signal = None + + if buy_signal: + if self._in_stop_cooldown(symbol): + logger.info("MeanRev BUY SKIPPED: %s [%s] (stop-out cooldown %d min)", + symbol, buy_signal, self.cfg.stop_cooldown_minutes) + return + ok, reason = self._can_open_position(symbol, self._trade_value_usd(symbol), last["close"]) + if not ok: + logger.info("MeanRev BUY SKIPPED: %s [%s] (%s)", symbol, buy_signal, reason) + return + qty = self._order_quantity(symbol, last["close"]) + logger.info( + "MeanRev BUY: %s x%d [%s] RSI=%.1f, close=%.2f, BB_lower=%.2f", + symbol, qty, buy_signal, last["rsi"], last["close"], last["bb_lower"], + ) + await self._buy(symbol, contract, qty) + else: + entry = owned["entry_price"] + profit_pct = (last["close"] - entry) / entry * 100 + + # stop loss always honoured, overrides min profit; + # soft stop is only a fallback: the exchange-side STP order is primary + if not self._has_active_stop(symbol) and last["close"] <= entry * (1 - self.cfg.stop_loss_pct / 100): + logger.info( + "MeanRev STOP-LOSS SELL (soft fallback): %s close=%.2f entry=%.2f (%.2f%%)", + symbol, last["close"], entry, profit_pct, + ) + await self._sell(symbol, contract, owned["quantity"]) + self._mark_stop_cooldown(symbol) + return + + sell_signal = None + if last["rsi"] > self.cfg.rsi_overbought: + sell_signal = "RSI" + elif last["close"] >= last["bb_mid"]: + sell_signal = "Bollinger" + elif profit_pct >= self.cfg.recovery_pct: + sell_signal = "Recovery" + + if sell_signal and profit_pct < self.cfg.min_profit_pct: + logger.debug( + "MeanRev SELL WAITING: %s [%s] profit=%.2f%% < min %.2f%%", + symbol, sell_signal, profit_pct, self.cfg.min_profit_pct, + ) + sell_signal = None + + if sell_signal: + logger.info( + "MeanRev SELL: %s [%s] RSI=%.1f, close=%.2f, entry=%.2f, profit=%.2f%%", + symbol, sell_signal, last["rsi"], last["close"], entry, profit_pct, + ) + await self._sell(symbol, contract, owned["quantity"]) + + async def _buy(self, symbol: str, contract: Contract, quantity: int): + ref = f"{self.name}:{symbol}" + trade = await execute_market_order(self.ib, contract, "BUY", quantity, ref) + if trade and trade.orderStatus.filled > 0: + self.tracker.record_buy( + self.name, symbol, trade.orderStatus.filled, trade.orderStatus.avgFillPrice + ) + if self._use_hard_stop(): + stop_price = self._target_stop_price( + symbol, trade.orderStatus.avgFillPrice, None + ) + await self._place_stop( + symbol, contract, trade.orderStatus.filled, stop_price + ) + else: + # order rejected/timed out (e.g. insufficient buying power): + # cool down to avoid retrying every cycle + self._mark_cooldown(symbol, "order failed") + + async def _sell(self, symbol: str, contract: Contract, quantity: float): + # cancel the hard stop first; it may have filled in the cancel race + if self._use_hard_stop(): + stop_filled, stop_price, cancel_confirmed = await self._cancel_stop(symbol) + if stop_filled > 0: + logger.info("%s: stop order filled %g during cancel", symbol, stop_filled) + self.tracker.record_sell(self.name, symbol, stop_filled, stop_price) + quantity -= stop_filled + if quantity <= 0: + return + if not cancel_confirmed: + # STP may still be live - a market sell could double-sell; + # leave the position for the next cycle + logger.warning( + "MeanRev SELL ABORTED: %s (stop cancel not confirmed - " + "may still be active, will retry next cycle)", symbol, + ) + return + ref = f"{self.name}:{symbol}" + trade = await execute_market_order(self.ib, contract, "SELL", quantity, ref) + if trade and trade.orderStatus.filled > 0: + self.tracker.record_sell(self.name, symbol, trade.orderStatus.filled, trade.orderStatus.avgFillPrice) + + async def on_tick(self): + pass + + def _contract(self, symbol: str): + return Contract( + symbol=symbol, + secType="STK", + exchange=self.cfg.exchange, + currency=self.cfg.currency, + ) diff --git a/strategies/short_term.py b/strategies/short_term.py new file mode 100644 index 0000000..7c5cc65 --- /dev/null +++ b/strategies/short_term.py @@ -0,0 +1,216 @@ +import logging +from datetime import date + +import pandas as pd +from ib_insync import Contract, IB + +from bars import BarManager, is_market_active, parse_bar_size_seconds, to_completed_df +from config import config +from orders import execute_market_order +from state import PositionTracker +from strategies.base import BaseStrategy + +logger = logging.getLogger(__name__) + + +class ShortTermMAVWAPStrategy(BaseStrategy): + """Fast EMA cross above slow EMA with price above daily VWAP. + + Exits (re-evaluated every cycle while in position): + - hard stop loss at -stop_loss_pct (always honoured) + - max holding period reached + - fast EMA below slow EMA, once profit >= min_profit_pct + """ + + def __init__(self, ib: IB, bar_manager: BarManager, tracker: PositionTracker): + super().__init__(ib, bar_manager, tracker) + self.cfg = config.short_term + self.fast_period = self.cfg.fast_ma_period + self.slow_period = self.cfg.slow_ma_period + self.bar_seconds = parse_bar_size_seconds(self.cfg.bar_size) + self.contracts: dict[str, Contract] = {} + + async def on_start(self): + hold_desc = "unlimited" if self.cfg.max_hold_days <= 0 else f"{self.cfg.max_hold_days} days" + logger.info( + "Short-term MA+VWAP strategy started: symbols=%s, fast=%d, slow=%d, value=$%.0f, " + "max_hold=%s, stop_loss=%.1f%%, min_profit=%.1f%%, confirm_bars=%d", + self.cfg.symbols, self.fast_period, self.slow_period, self.cfg.trade_value_usd, + hold_desc, self.cfg.stop_loss_pct, self.cfg.min_profit_pct, + self.cfg.entry_confirm_bars, + ) + for symbol in self.cfg.symbols: + contract = self._contract(symbol) + await self.ib.qualifyContractsAsync(contract) + if not contract.conId: + logger.error("ShortTerm %s: failed to qualify contract, skipped", symbol) + continue + self.contracts[symbol] = contract + await self.bar_manager.subscribe( + contract, self.cfg.bar_size, f"{self.cfg.lookback_days} D", "TRADES" + ) + logger.info("ShortTerm %s: subscribed, owned=%s", symbol, self.tracker.get(self.name, symbol)) + + @staticmethod + def _calc_vwap(df: pd.DataFrame) -> pd.Series: + """VWAP reset each trading day (vectorized, no groupby-apply).""" + typical_price = (df["high"] + df["low"] + df["close"]) / 3 + pv = typical_price * df["volume"] + day = df["date"].dt.date + cum_pv = pv.groupby(day).cumsum() + cum_vol = df["volume"].groupby(day).cumsum() + return cum_pv / cum_vol.replace(0, float("nan")) + + @staticmethod + def _calc_ema(series: pd.Series, period: int) -> pd.Series: + return series.ewm(span=period, adjust=False).mean() + + async def _get_df(self, contract: Contract) -> pd.DataFrame: + bars = self.bar_manager.get(contract, self.cfg.bar_size, "TRADES") + if bars is None: + bars = await self.bar_manager.subscribe( + contract, self.cfg.bar_size, f"{self.cfg.lookback_days} D", "TRADES" + ) + return to_completed_df(bars, self.bar_seconds) + + async def on_bar(self): + await self._sync_stop_orders() + for symbol, contract in self.contracts.items(): + try: + await self._process_symbol(symbol, contract) + except Exception as e: + logger.exception("ShortTerm %s: error: %s", symbol, e) + + async def _process_symbol(self, symbol: str, contract: Contract): + df = await self._get_df(contract) + if len(df) < self.slow_period + 2: + return + if not is_market_active(df, self.bar_seconds): + return # market closed / stale data - never trade + + df = df.copy() + df["fast_ema"] = self._calc_ema(df["close"], self.fast_period) + df["slow_ema"] = self._calc_ema(df["close"], self.slow_period) + df["vwap"] = self._calc_vwap(df) + + last = df.iloc[-1] + prev = df.iloc[-2] + fast_above = last["fast_ema"] > last["slow_ema"] + + # entry confirmation: the cross happened (confirm_bars-1) bars back and + # the fast EMA is still above the slow EMA now -> filters 1-bar whipsaws + cb = self.cfg.entry_confirm_bars + ref = df.iloc[-cb] + ref_prev = df.iloc[-cb - 1] + cross_then = ref["fast_ema"] > ref["slow_ema"] and ref_prev["fast_ema"] <= ref_prev["slow_ema"] + confirmed = cross_then and fast_above + + price_above_vwap = last["close"] > last["vwap"] + + owned = self.tracker.get(self.name, symbol) + + if owned: + entry = owned["entry_price"] + profit_pct = (last["close"] - entry) / entry * 100 + + # soft stop is only a fallback: the exchange-side STP order is primary + if not self._has_active_stop(symbol) and last["close"] <= entry * (1 - self.cfg.stop_loss_pct / 100): + logger.info( + "ShortTerm STOP-LOSS SELL (soft fallback): %s close=%.2f entry=%.2f (%.2f%%)", + symbol, last["close"], entry, profit_pct, + ) + await self._sell(symbol, contract, owned["quantity"]) + self._mark_stop_cooldown(symbol) + return + + entry_date = date.fromisoformat(owned["entry_date"]) if owned.get("entry_date") else None + if entry_date and self.cfg.max_hold_days > 0: + hold_days = (date.today() - entry_date).days + if hold_days >= self.cfg.max_hold_days: + logger.info("ShortTerm SELL %s: max hold reached (%d days)", symbol, hold_days) + await self._sell(symbol, contract, owned["quantity"]) + return + + if not fast_above: + # re-checked every cycle: exits as soon as profit requirement is met + if profit_pct >= self.cfg.min_profit_pct: + logger.info( + "ShortTerm SELL: %s (fast EMA %.2f < slow EMA %.2f, profit=%.2f%%)", + symbol, last["fast_ema"], last["slow_ema"], profit_pct, + ) + await self._sell(symbol, contract, owned["quantity"]) + else: + logger.debug( + "ShortTerm SELL WAITING: %s profit %.2f%% < min %.1f%%", + symbol, profit_pct, self.cfg.min_profit_pct, + ) + else: + if confirmed and price_above_vwap: + if self._in_stop_cooldown(symbol): + logger.info("ShortTerm BUY SKIPPED: %s (stop-out cooldown %d min)", + symbol, self.cfg.stop_cooldown_minutes) + return + ok, reason = self._can_open_position(symbol, self._trade_value_usd(symbol), last["close"]) + if not ok: + logger.info("ShortTerm BUY SKIPPED: %s (%s)", symbol, reason) + return + qty = self._order_quantity(symbol, last["close"]) + logger.info( + "ShortTerm BUY: %s x%d (cross confirmed over %d bars, fast EMA %.2f > slow EMA %.2f, close %.2f > VWAP %.2f)", + symbol, qty, self.cfg.entry_confirm_bars, + last["fast_ema"], last["slow_ema"], last["close"], last["vwap"], + ) + await self._buy(symbol, contract, qty) + + async def _buy(self, symbol: str, contract: Contract, quantity: int): + ref = f"{self.name}:{symbol}" + trade = await execute_market_order(self.ib, contract, "BUY", quantity, ref) + if trade and trade.orderStatus.filled > 0: + self.tracker.record_buy( + self.name, symbol, trade.orderStatus.filled, trade.orderStatus.avgFillPrice + ) + if self._use_hard_stop(): + stop_price = self._target_stop_price( + symbol, trade.orderStatus.avgFillPrice, None + ) + await self._place_stop( + symbol, contract, trade.orderStatus.filled, stop_price + ) + else: + # order rejected/timed out (e.g. insufficient buying power): + # cool down to avoid retrying every cycle + self._mark_cooldown(symbol, "order failed") + + async def _sell(self, symbol: str, contract: Contract, quantity: float): + # cancel the hard stop first; it may have filled in the cancel race + if self._use_hard_stop(): + stop_filled, stop_price, cancel_confirmed = await self._cancel_stop(symbol) + if stop_filled > 0: + logger.info("%s: stop order filled %g during cancel", symbol, stop_filled) + self.tracker.record_sell(self.name, symbol, stop_filled, stop_price) + quantity -= stop_filled + if quantity <= 0: + return + if not cancel_confirmed: + # STP may still be live - a market sell could double-sell; + # leave the position for the next cycle + logger.warning( + "ShortTerm SELL ABORTED: %s (stop cancel not confirmed - " + "may still be active, will retry next cycle)", symbol, + ) + return + ref = f"{self.name}:{symbol}" + trade = await execute_market_order(self.ib, contract, "SELL", quantity, ref) + if trade and trade.orderStatus.filled > 0: + self.tracker.record_sell(self.name, symbol, trade.orderStatus.filled, trade.orderStatus.avgFillPrice) + + async def on_tick(self): + pass + + def _contract(self, symbol: str): + return Contract( + symbol=symbol, + secType="STK", + exchange=self.cfg.exchange, + currency=self.cfg.currency, + )