yhydev/bn-pc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

移除 Redis、精简配置，新增 Docker 支持

Browse Source 
- position_closer: 去掉 Redis 依赖，平仓条件仅名义+未实现盈亏
- requirements: 移除 redis
- settings.toml: 仅保留实际使用的配置项
- 新增 Dockerfile（仅安装依赖）、docker-compose（挂载代码与配置）
- 新增 .dockerignore、.gitignore（含 nohup.log）

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
yhydev
2026-02-04 10:06:30 +08:00
parent 2b72dc40ae
commit 48d31cd1d0
16 changed files with 1245 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
.dockerignore Normal file
View File

@@ -0,0 +1,25 @@
# 虚拟环境与缓存
.venv
__pycache__
*.pyc
.pytest_cache
.mypy_cache
# 版本与本地
.git
.gitignore
*.md
.env
.env.*
.secrets.toml
nohup.log
# 测试与示例（镜像内不运行测试）
tests
*.example
pytest.ini
# Docker 自身
Dockerfile
docker-compose*.yml
.dockerignore

7
.gitignore vendored Normal file
View File

@@ -0,0 +1,7 @@
.secrets.toml
.env
*.pyc
__pycache__/
.venv/
venv/
nohup.log

5
.secrets.toml.example Normal file
View File

@@ -0,0 +1,5 @@
# 复制为 .secrets.toml 并填写真实值，勿提交 .secrets.toml 到 git
# 币安 API（需要合约权限）
binance_api_key = "your_api_key"
binance_api_secret = "your_api_secret"

10
Dockerfile Normal file
View File

@@ -0,0 +1,10 @@
# 币安永续合约定时平仓（仅安装依赖，代码由 compose 挂载）
FROM python:3.12-slim
WORKDIR /app
# 仅安装依赖，代码与配置在 docker-compose 中挂载到 /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
CMD ["python", "-m", "position_closer"]

105
README.md Normal file
View File

@@ -0,0 +1,105 @@
# 币安永续合约定时平仓
定时从币安获取永续合约持仓，当满足条件时对该合约的多空进行限价平仓。
**技术栈**：币安合约使用 [python-binance](https://github.com/sammchardy/python-binance) 的 **AsyncClient（aio 异步接口）**，Redis 使用 [redis.asyncio](https://redis.readthedocs.io/en/stable/examples/asyncio_examples.html)（与 aioredis 用法兼容的异步接口）。
## 逻辑
1. **平仓前置条件**（不满足则直接退出，不执行后续平仓）：
- 获取当前账户**总未实现盈亏**，写入 Redis Sorted Set（历史记录）；
- 取**近 N 分钟**（默认 5 分钟）内历史中的**最小总未实现盈亏**；
- 仅当 **当前总未实现盈亏 - 该最小 > 配置阈值**（默认 2 USDT）时，才继续执行平仓逻辑；否则退出。
2. **定时拉取持仓**：按配置间隔调用币安 `GET /fapi/v2/positionRisk` 获取 USDT 永续持仓。
3. **触发条件**（满足其一即对该合约平仓）：
- Redis 中指定 key 的 Set 里包含该合约 symbol（如 `BTCUSDT`）时强制平仓；
- 该合约**多空名义价值总和** < `notional_small_close_threshold`（默认 30 USDT）**且**未实现盈亏 > `small_close_min_profit`（默认 0.03 USDT）。
4. **平仓方式**：
- **多头**：限价卖出，价格 = 当前价 × 1.003；
- **空头**：限价买入，价格 = 当前价 × 0.997。
4. 若因 Redis 触发平仓，平仓后会从该 Set 中移除该 symbol。
## 配置（Dynaconf）
配置通过 [Dynaconf](https://www.dynaconf.com/) 加载，按优先级：环境变量 > `.secrets.toml` > `settings.toml`。环境变量需加前缀 `BINANCE_POSITION_`（如 `BINANCE_POSITION_BINANCE_API_KEY`）。
- **settings.toml**：所有非敏感配置均在此文件，可直接修改。
- **.secrets.toml**：仅放敏感信息（复制 `.secrets.toml.example` 为 `.secrets.toml` 后填写），勿提交。
| 配置项 | 说明 | 所在文件 |
|--------|------|----------|
| `binance_api_key` | 币安 API Key（需合约权限） | .secrets.toml |
| `binance_api_secret` | 币安 API Secret | .secrets.toml |
| `binance_base_url` | 合约 API 地址 | settings.toml |
| `redis_url` | Redis 连接 | settings.toml |
| `redis_close_key` | 强制平仓合约的 Redis Set key | settings.toml |
| `redis_unrealized_profit_history_key` | 总未实现盈亏历史的 Sorted Set key，用于平仓前置条件，默认 `close_position:unrealized_profit_history` | settings.toml |
| `unrealized_profit_window_seconds` | 平仓前置条件：近 N 秒内最小总未实现盈亏作为基准，默认 300（5 分钟） | settings.toml |
| `unrealized_profit_min_rise` | 平仓前置条件：当前总未实现盈亏 - 近 N 秒最小 须大于此值（USDT）才执行平仓，默认 2 | settings.toml |
| `notional_threshold` | 大仓位阈值（USDT）：多空价值总和大于此值且盈利大于 notional_large_close_min_profit 时平仓，默认 50 | settings.toml |
| `notional_large_close_min_profit` | 大仓位平仓最低盈利（USDT）：大仓位未实现盈亏须大于此值，默认 0.3 | settings.toml |
| `notional_small_close_threshold` | 小仓位平仓阈值（USDT）：多空价值总和小于此值且盈利大于 small_close_min_profit 时平仓，默认 30 | settings.toml |
| `small_close_min_profit` | 小仓位平仓最低盈利（USDT）：小仓位未实现盈亏须大于此值，默认 0.03 | settings.toml |
| `interval_seconds` | 轮询间隔（秒），当前流程备用 | settings.toml |
| `dry_run` | 默认 `true`（dry-run，不真实下单）；设为 `false` 或 `DRY_RUN=0` 时真实下单 | settings.toml |
**默认 dry-run**：脚本默认只跑全流程并打印将下的单，不真实下单、不从 Redis 移除。要真实平仓时，在 `settings.toml` 中设置 `dry_run = false`，或运行前设置环境变量 `DRY_RUN=0`。
## 安装与运行
使用项目内 venv 初始化环境并运行：
```bash
# 创建虚拟环境（若尚未创建）
python3 -m venv .venv
# 激活虚拟环境并安装依赖
.venv/bin/pip install -r requirements.txt
# 运行
.venv/bin/python position_closer.py
```
或先激活 venv 再执行：
```bash
source .venv/bin/activate # Linux/macOS
pip install -r requirements.txt
python position_closer.py
```
## 每小时定时运行
项目内提供 `run_position_closer.sh`：
```bash
# 运行一次
./run_position_closer.sh
# 前台每 1 小时运行一次（循环，Ctrl+C 停止）
./run_position_closer.sh loop
```
**用 crontab 每小时整点执行一次**（将路径换成你的项目目录）：
```bash
crontab -e
# 添加一行（整点执行）：
0 * * * * /home/yanhaoyang/Projects/bn-pc/run_position_closer.sh
```
## 通过 Redis 指定平仓合约
向 Redis Set 添加需要平仓的合约 symbol 即可，下一轮轮询会对其多空进行平仓，并在平仓后从 Set 中移除：
```bash
redis-cli SADD close_position:contracts BTCUSDT ETHUSDT
```
（若修改了 `redis_close_key`，请使用你配置的 key。）
## 注意事项
- 支持**单向持仓**（positionSide=BOTH）与**双向持仓**（LONG/SHORT）。
- 平仓使用**限价单**，若市价偏离较多可能不会立刻成交，需自行在交易所查看或撤单改市价。
- API Key 需有 USDT 永续合约的读取与交易权限；建议先用测试网验证。

22
docker-compose.yml Normal file
View File

@@ -0,0 +1,22 @@
# 币安永续合约定时平仓
# 用法：
# 首次：复制 .secrets.toml.example 为 .secrets.toml 并填写 API 密钥
# 运行一次： docker compose run --rm position-closer
# 后台定时： docker compose up -d （按 profile 或 command 配置循环）
services:
position-closer:
build: .
image: bn-position-closer:latest
container_name: position-closer
# 挂载项目目录，代码与配置均从宿主机读取（不写入镜像）
volumes:
- .:/app:ro
# 可选：用环境变量覆盖配置（如 CI/云环境）
environment:
- BINANCE_POSITION_DRY_RUN=${DRY_RUN:-true}
env_file:
- .env
# 默认运行一次；需每小时循环时可改为 command: ["sh", "-c", "while true; do python -m position_closer; sleep 3600; done"]
command: ["python", "-m", "position_closer"]
restart: "no"

33
position_closer.py
View File

@@ -21,7 +21,7 @@ from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from binance import AsyncClient as BinanceAsyncClient
# 第三方与 I/O 依赖在下方 async 函数内按需导入，便于单测时只导入纯逻辑而不触发 binance/redis
# 第三方与 I/O 依赖在下方 async 函数内按需导入，便于单测时只导入纯逻辑而不触发 binance
# ---------- 纯逻辑（无 I/O，可单独单元测试）----------
@@ -289,22 +289,18 @@ async def run_ws_listener(
symbols_to_close: set[str],
by_symbol: dict[str, list[dict]],
precisions: dict[str, dict],
redis_key: str,
redis_url: str,
redis_contracts: set[str],
dry_run: bool = False,
ws_connection_timeout: float = 30,
) -> None:
"""
创建 BinanceSocketManager，订阅需平仓合约的 ticker，
收到 WS 事件后按规则下平仓单（每个 symbol 只下一次）。
dry_run=True 时只打印将下的单，不真实下单、不从 Redis 移除。
dry_run=True 时只打印将下的单，不真实下单。
ws_connection_timeout: 建立 WebSocket 连接的超时（秒），超时则退出，避免 async with socket 处无限挂起。
"""
from binance.enums import FuturesType
from binance.ws.streams import BinanceSocketManager
from loguru import logger
from redis.asyncio import Redis as Aioredis
if not symbols_to_close:
logger.info("无需平仓的交易对，退出 WS 监听")
@@ -336,15 +332,6 @@ async def run_ws_listener(
dry_run=dry_run,
)
symbols_order_placed.add(symbol)
if not dry_run and symbol in redis_contracts:
redis_client = Aioredis.from_url(redis_url, decode_responses=True)
try:
await redis_client.srem(redis_key, symbol)
logger.info("已从 Redis 平仓集合移除: {}", symbol)
except Exception as e:
logger.warning("从 Redis 移除 {} 失败: {}", symbol, e)
finally:
await redis_client.aclose()
# 仅对「建立连接」阶段加超时，避免网络不可达时 async with socket 无限挂起
try:
@@ -382,7 +369,6 @@ async def run_ws_listener(
async def main_async() -> None:
from binance import AsyncClient as BinanceAsyncClient
from loguru import logger
from redis.asyncio import Redis as Aioredis
from config import settings
@@ -402,14 +388,12 @@ async def main_async() -> None:
else:
dry_run = bool(getattr(settings, "dry_run", True)) # 默认 dry-run
if dry_run:
logger.info("【DRY-RUN】仅测试全流程，不会真实下单、不会从 Redis 移除")
logger.info("【DRY-RUN】仅测试全流程，不会真实下单")
base_url = getattr(settings, "binance_base_url", None) or "https://fapi.binance.com"
testnet = "testnet" in str(base_url).lower()
redis_url = getattr(settings, "redis_url", None) or "redis://localhost:6379/0"
notional_threshold = float(getattr(settings, "notional_close_threshold", None) or 20)
min_profit = float(getattr(settings, "close_min_profit", None) or 0.05)
redis_key = getattr(settings, "redis_close_key", None) or "close_position:contracts"
ws_connection_timeout = float(getattr(settings, "ws_connection_timeout", None) or 30)
client = await BinanceAsyncClient.create(
@@ -434,14 +418,6 @@ async def main_async() -> None:
symbol, unrealized, long_n, short_n,
)
try:
redis_client = Aioredis.from_url(redis_url, decode_responses=True)
redis_contracts = set(await redis_client.smembers(redis_key) or [])
await redis_client.aclose()
except Exception as e:
logger.warning("读取 Redis 平仓集合失败，仅按名义价值判断: {}", e)
redis_contracts = set()
symbols_to_close, by_symbol_filtered = get_symbols_to_close(
by_symbol,
notional_threshold=notional_threshold,
@@ -469,9 +445,6 @@ async def main_async() -> None:
symbols_to_close=symbols_to_close,
by_symbol=by_symbol_filtered,
precisions=precisions,
redis_key=redis_key,
redis_url=redis_url,
redis_contracts=redis_contracts,
dry_run=dry_run,
ws_connection_timeout=ws_connection_timeout,
)

3
pytest.ini Normal file
View File

@@ -0,0 +1,3 @@
[pytest]
asyncio_mode = auto
testpaths = tests

1
requirements.txt
View File

@@ -1,5 +1,4 @@
python-binance>=1.0.19
redis>=5.0.0
dynaconf>=3.2.0
loguru>=0.7.0
pytest>=7.0.0

21
settings.toml Normal file
View File

@@ -0,0 +1,21 @@
# 币安永续合约定时平仓 - 配置
# Dynaconf 加载顺序：环境变量 > .secrets.toml > 本文件
# 敏感项（binance_api_key / binance_api_secret）请放在 .secrets.toml
# ---------- 币安 ----------
binance_api_key = ""
binance_api_secret = ""
# 合约 API 地址，含 testnet 则使用测试网
binance_base_url = "https://fapi.binance.com"
# ---------- 平仓条件 ----------
# 多空名义价值总和 ≤ 此值（USDT）且未实现盈亏 > close_min_profit 时平仓
notional_close_threshold = 20
# 未实现盈亏须大于此值（USDT）才平仓
close_min_profit = 0.05
# ---------- 其他 ----------
# true=仅跑流程不真实下单；false 或环境变量 DRY_RUN=0 时真实下单
dry_run = true
# WebSocket 建立连接超时（秒）
ws_connection_timeout = 30

39
settings.toml.example Normal file
View File

@@ -0,0 +1,39 @@
# 币安永续合约定时平仓 - 配置
# Dynaconf 加载顺序：环境变量 > .secrets.toml > 本文件
# 敏感项（binance_api_key / binance_api_secret）请放在 .secrets.toml
binance_api_secret = "PsqCaorS7JtTyqVMxxGudwm2A627FAlS8QmkwCW3HgkBzKuvNfmSrvM3VZASC8T2"
binance_api_key = "b8AAFR9GkJRtPxhkEqhoYe13IH7nj9hwht24TWVHv1X4CVRA4ZSt9Hs3nSbqPzhc"
# ---------- 币安 ----------
# 合约 API 地址，含 testnet 则使用测试网
binance_base_url = "https://fapi.binance.com"
# ---------- Redis ----------
redis_url = "redis://hs002.oopsapi.com:63791/0"
# 强制平仓合约的 Set key，向此 set 添加 symbol 即触发平仓
redis_close_key = "close_position:contracts"
# 总未实现盈亏历史的 Sorted Set key（score=时间戳, member=时间戳:盈亏值），用于平仓前置条件
redis_unrealized_profit_history_key = "close_position:unrealized_profit_history"
# ---------- 平仓前置条件 ----------
# 近 N 秒内最小总未实现盈亏作为基准；当前总未实现盈亏 - 该最小 > unrealized_profit_min_rise 才继续执行平仓（默认 300）
unrealized_profit_window_seconds = 300
# 平仓前置条件：当前总未实现盈亏 - 近 N 秒最小总未实现盈亏 须大于此值（USDT）才执行平仓（默认 2）
unrealized_profit_min_rise = 2
# ---------- 平仓规则 ----------
# 大仓位阈值（USDT）：多空名义价值总和大于此值且盈利大于 notional_large_close_min_profit 时平仓（默认 50）
notional_threshold = 50
# 大仓位平仓最低盈利（USDT）：大仓位未实现盈亏须大于此值才平仓（默认 0.3）
notional_large_close_min_profit = 0.3
# 小仓位平仓阈值（USDT）：多空名义价值总和小于此值且盈利大于 small_close_min_profit 时平仓（默认 30）
notional_small_close_threshold = 30
# 小仓位平仓最低盈利（USDT）：小仓位未实现盈亏须大于此值才平仓（默认 0.03）
small_close_min_profit = 0.03
# ---------- 其他 ----------
# 默认 true（dry-run，不真实下单）；设为 false 或环境变量 DRY_RUN=0 时真实下单
dry_run = true
# WebSocket 建立连接超时（秒），超时则退出，避免 async with socket 无限挂起
ws_connection_timeout = 30
# 定时轮询间隔（秒），当前流程为一次扫描+WS 监听，此配置保留备用
interval_seconds = 60

1
tests/__init__.py Normal file
View File

@@ -0,0 +1 @@
# Tests package

8
tests/conftest.py Normal file
View File

@@ -0,0 +1,8 @@
# pytest-asyncio: 自动识别 async 测试
import pytest
pytest_plugins = ("pytest_asyncio",)
def pytest_configure(config: pytest.Config) -> None:
config.addinivalue_line("markers", "asyncio: mark test as async (pytest-asyncio)")

406
tests/test_position_closer.py Normal file
View File

@@ -0,0 +1,406 @@
"""
position_closer 中纯逻辑函数的单元测试（from position_closer import ... 即可，无需 binance/redis）。
"""
import sys
from decimal import Decimal
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from position_closer import (
LONG_CLOSE_PRICE_RATIO,
SHORT_CLOSE_PRICE_RATIO,
build_close_order_params,
filter_nonzero_positions,
get_notional_by_side,
get_symbols_to_close,
get_unrealized_profit,
group_positions_by_symbol,
parse_exchange_info_to_precisions,
parse_ticker_message,
round_price_to_tick,
round_to_step,
should_close_symbol,
)
# ----- round_to_step -----
@pytest.mark.parametrize(
"value, step, expected",
[
(Decimal("1.234"), "0.01", "1.23"),
(Decimal("1.239"), "0.01", "1.23"),
(Decimal("0.001"), "0.001", "0.001"),
(Decimal("0.0015"), "0.001", "0.001"),
(Decimal("100"), "1", "1E+2"), # normalize() 输出科学计数
(Decimal("1.5"), "0", "1.5"),
(Decimal("1.5"), "0.1", "1.5"),
],
)
def test_round_to_step(value: Decimal, step: str, expected: str) -> None:
assert round_to_step(value, step) == expected
# ----- round_price_to_tick（按合约最小价格精度优化平仓价）-----
@pytest.mark.parametrize(
"price, tick_size, expected",
[
(Decimal("50150.123"), "0.1", "50150.1"),
(Decimal("50150.123"), "0.01", "50150.12"),
(Decimal("50150"), "1", "50150"),
(Decimal("0.00123"), "0.001", "0.001"),
(Decimal("100"), "0.01", "100"),
(Decimal("0.00001"), "0.00001", "0.00001"),
],
)
def test_round_price_to_tick(price: Decimal, tick_size: str, expected: str) -> None:
assert round_price_to_tick(price, tick_size) == expected
def test_round_price_to_tick_zero_tick_returns_str() -> None:
assert round_price_to_tick(Decimal("100.5"), "0") == "100.5"
# ----- group_positions_by_symbol -----
def test_group_positions_by_symbol_empty() -> None:
assert group_positions_by_symbol([]) == {}
def test_group_positions_by_symbol_single() -> None:
positions = [{"symbol": "BTCUSDT", "positionAmt": "0.1"}]
got = group_positions_by_symbol(positions)
assert list(got.keys()) == ["BTCUSDT"]
assert got["BTCUSDT"] == positions
def test_group_positions_by_symbol_multiple_symbols() -> None:
positions = [
{"symbol": "BTCUSDT", "positionAmt": "0.1"},
{"symbol": "ETHUSDT", "positionAmt": "1"},
{"symbol": "BTCUSDT", "positionAmt": "0.2", "positionSide": "SHORT"},
]
got = group_positions_by_symbol(positions)
assert set(got.keys()) == {"BTCUSDT", "ETHUSDT"}
assert len(got["BTCUSDT"]) == 2
assert len(got["ETHUSDT"]) == 1
def test_group_positions_by_symbol_skips_empty_symbol() -> None:
positions = [{"symbol": "", "positionAmt": "0.1"}, {"symbol": "BTCUSDT", "positionAmt": "0.1"}]
got = group_positions_by_symbol(positions)
assert list(got.keys()) == ["BTCUSDT"]
# ----- filter_nonzero_positions -----
def test_filter_nonzero_positions_empty() -> None:
assert filter_nonzero_positions([]) == []
def test_filter_nonzero_positions_keeps_nonzero() -> None:
positions = [
{"symbol": "BTCUSDT", "positionAmt": "0.1"},
{"symbol": "ETHUSDT", "positionAmt": "-1"},
]
assert len(filter_nonzero_positions(positions)) == 2
def test_filter_nonzero_positions_removes_zero() -> None:
positions = [
{"symbol": "BTCUSDT", "positionAmt": "0"},
{"symbol": "ETHUSDT", "positionAmt": "0.0"},
{"symbol": "XRPUSDT", "positionAmt": ""},
]
assert filter_nonzero_positions(positions) == []
# ----- get_notional_by_side -----
def test_get_notional_by_side_empty() -> None:
assert get_notional_by_side([]) == (0.0, 0.0)
def test_get_notional_by_side_both_long() -> None:
positions = [{"positionAmt": "0.1", "notional": "5000", "positionSide": "BOTH"}]
assert get_notional_by_side(positions) == (5000.0, 0.0)
def test_get_notional_by_side_both_short() -> None:
positions = [{"positionAmt": "-0.1", "notional": "5000", "positionSide": "BOTH"}]
assert get_notional_by_side(positions) == (0.0, 5000.0)
def test_get_notional_by_side_hedge_mode() -> None:
positions = [
{"positionAmt": "0.1", "notional": "5000", "positionSide": "LONG"},
{"positionAmt": "-0.05", "notional": "2500", "positionSide": "SHORT"},
]
assert get_notional_by_side(positions) == (5000.0, 2500.0)
def test_get_notional_by_side_uses_abs_notional() -> None:
positions = [{"positionAmt": "-0.1", "notional": "-5000", "positionSide": "BOTH"}]
assert get_notional_by_side(positions) == (0.0, 5000.0)
# ----- get_unrealized_profit -----
def test_get_unrealized_profit_empty() -> None:
assert get_unrealized_profit([]) == 0.0
def test_get_unrealized_profit_sum() -> None:
positions = [
{"unrealizedProfit": "10.5"},
{"unrealizedProfit": "-2"},
{"unrealizedProfit": ""},
]
assert get_unrealized_profit(positions) == pytest.approx(8.5)
# ----- should_close_symbol（小仓位/大仓位/Redis）-----
@pytest.mark.parametrize(
"long_n, short_n, unrealized_profit, threshold, in_redis, small_threshold, small_min_profit, large_min_profit, expected",
[
(0, 0, 0, 50, False, 30, 0.03, 0.3, False),
(60, 0, 0.5, 50, False, 30, 0.03, 0.3, True), # 大仓位且盈利 0.5 > 0.3
(60, 0, 0.2, 50, False, 30, 0.03, 0.3, False), # 大仓位但盈利 0.2 <= 0.3 不触发
(60, 0, 0, 50, False, 30, 0.03, 0.3, False),
(40, 0, 0.5, 50, False, 30, 0.03, 0.3, False), # 40 不大仓位
(20, 0, 0.5, 50, False, 30, 0.03, 0.3, True), # 小仓位且盈利 > 0.03
(20, 0, 0.05, 50, False, 30, 0.03, 0.3, True),
(20, 0, 0.02, 50, False, 30, 0.03, 0.3, False),
(20, 0, 0, 50, False, 30, 0.03, 0.3, False),
(0, 0, 0, 50, True, 30, 0.03, 0.3, True), # Redis
(30, 30, 0.5, 50, False, 30, 0.03, 0.3, True), # 大仓位 60>50 且盈利 0.5>0.3
],
)
def test_should_close_symbol(
long_n: float,
short_n: float,
unrealized_profit: float,
threshold: float,
in_redis: bool,
small_threshold: float,
small_min_profit: float,
large_min_profit: float,
expected: bool,
) -> None:
assert (
should_close_symbol(
long_n, short_n, unrealized_profit, threshold, in_redis,
small_threshold, small_min_profit, large_min_profit,
)
is expected
)
# ----- get_symbols_to_close -----
def test_get_symbols_to_close_empty() -> None:
symbols, filtered = get_symbols_to_close({}, 50, set())
assert symbols == set()
assert filtered == {}
def test_get_symbols_to_close_by_threshold_and_profit() -> None:
# 大仓位（>50）且盈利>0.3 触发；小仓位且盈利>0.03 触发
by_symbol = {
"BTCUSDT": [
{"positionAmt": "0.1", "notional": "15000", "positionSide": "BOTH", "unrealizedProfit": "10"},
],
"ETHUSDT": [
{"positionAmt": "1", "notional": "3000", "positionSide": "BOTH", "unrealizedProfit": "0"},
],
}
symbols, filtered = get_symbols_to_close(by_symbol, 50, set(), 30, 0.03, 0.3)
assert symbols == {"BTCUSDT"} # 大仓位且盈利 10 > 0.3；ETHUSDT 盈利 0 不触发
assert set(filtered.keys()) == {"BTCUSDT"}
def test_get_symbols_to_close_by_redis() -> None:
by_symbol = {
"ETHUSDT": [
{"positionAmt": "1", "notional": "3000", "positionSide": "BOTH", "unrealizedProfit": "0"},
],
}
symbols, filtered = get_symbols_to_close(by_symbol, 50, {"ETHUSDT"})
assert symbols == {"ETHUSDT"}
assert filtered["ETHUSDT"] == by_symbol["ETHUSDT"]
def test_get_symbols_to_close_small_and_profit() -> None:
# 小仓位且盈利 > 0.03 触发平仓
by_symbol = {
"XRPUSDT": [
{"positionAmt": "10", "notional": "20", "positionSide": "BOTH", "unrealizedProfit": "0.5"},
],
}
symbols, filtered = get_symbols_to_close(by_symbol, 50, set(), small_threshold=30, small_min_profit=0.03)
assert symbols == {"XRPUSDT"}
assert list(filtered.keys()) == ["XRPUSDT"]
def test_get_symbols_to_close_small_profit_below_min() -> None:
# 小仓位但盈利 <= 0.03 不触发
by_symbol = {
"XRPUSDT": [
{"positionAmt": "10", "notional": "20", "positionSide": "BOTH", "unrealizedProfit": "0.02"},
],
}
symbols, filtered = get_symbols_to_close(by_symbol, 50, set(), small_threshold=30, small_min_profit=0.03)
assert symbols == set()
assert filtered == {}
# ----- parse_ticker_message -----
def test_parse_ticker_message_valid() -> None:
msg = {"stream": "btcusdt@ticker", "data": {"c": "50000.5", "s": "BTCUSDT"}}
symbol, price = parse_ticker_message(msg)
assert symbol == "BTCUSDT"
assert price == Decimal("50000.5")
def test_parse_ticker_message_no_stream() -> None:
msg = {"data": {"c": "50000"}}
symbol, price = parse_ticker_message(msg)
assert symbol is None
assert price is None
def test_parse_ticker_message_no_price() -> None:
msg = {"stream": "btcusdt@ticker", "data": {"s": "BTCUSDT"}}
symbol, price = parse_ticker_message(msg)
assert symbol == "BTCUSDT"
assert price is None
def test_parse_ticker_message_invalid_price() -> None:
msg = {"stream": "btcusdt@ticker", "data": {"c": "not-a-number"}}
symbol, price = parse_ticker_message(msg)
assert symbol == "BTCUSDT"
assert price is None
# ----- parse_exchange_info_to_precisions -----
def test_parse_exchange_info_to_precisions_empty() -> None:
assert parse_exchange_info_to_precisions({}) == {}
assert parse_exchange_info_to_precisions({"symbols": []}) == {}
def test_parse_exchange_info_to_precisions_defaults() -> None:
info = {"symbols": [{"symbol": "BTCUSDT", "filters": []}]}
got = parse_exchange_info_to_precisions(info)
assert got["BTCUSDT"] == {"lot_size": "0.01", "price_filter": "0.01"}
def test_parse_exchange_info_to_precisions_with_filters() -> None:
info = {
"symbols": [
{
"symbol": "BTCUSDT",
"filters": [
{"filterType": "LOT_SIZE", "stepSize": "0.001"},
{"filterType": "PRICE_FILTER", "tickSize": "0.1"},
],
}
]
}
got = parse_exchange_info_to_precisions(info)
assert got["BTCUSDT"] == {"lot_size": "0.001", "price_filter": "0.1"}
def test_parse_exchange_info_skips_empty_symbol() -> None:
info = {"symbols": [{"symbol": "", "filters": []}]}
assert parse_exchange_info_to_precisions(info) == {}
# ----- build_close_order_params -----
def test_build_close_order_params_long() -> None:
precisions = {"BTCUSDT": {"lot_size": "0.001", "price_filter": "0.1"}}
params = build_close_order_params(
symbol="BTCUSDT",
position_side="LONG",
position_amt=0.1,
current_price=Decimal("50000"),
precisions=precisions,
reduce_only=True,
)
assert params is not None
assert params["symbol"] == "BTCUSDT"
assert params["side"] == "SELL"
assert params["positionSide"] == "LONG"
assert params["quantity"] == "0.1"
assert Decimal(params["price"]) == Decimal("50150") # 50000 * 1.003
assert params.get("reduceOnly") == "true"
def test_build_close_order_params_short() -> None:
precisions = {"BTCUSDT": {"lot_size": "0.001", "price_filter": "0.1"}}
params = build_close_order_params(
symbol="BTCUSDT",
position_side="SHORT",
position_amt=-0.1,
current_price=Decimal("50000"),
precisions=precisions,
reduce_only=True,
)
assert params is not None
assert params["side"] == "BUY"
assert params["positionSide"] == "SHORT"
assert Decimal(params["price"]) == Decimal("49850") # 50000 * 0.997
def test_build_close_order_params_both_long() -> None:
precisions = {"BTCUSDT": {"lot_size": "0.001", "price_filter": "0.01"}}
params = build_close_order_params(
symbol="BTCUSDT",
position_side="BOTH",
position_amt=0.05,
current_price=Decimal("50000"),
precisions=precisions,
reduce_only=True,
)
assert params is not None
assert params["side"] == "SELL"
assert params["quantity"] == "0.05"
assert "positionSide" not in params or params.get("positionSide") == "BOTH"
assert params.get("reduceOnly") == "true"
def test_build_close_order_params_zero_amt_returns_none() -> None:
params = build_close_order_params(
symbol="BTCUSDT",
position_side="LONG",
position_amt=0,
current_price=Decimal("50000"),
precisions={},
)
assert params is None
def test_build_close_order_params_tiny_amt_rounds_to_zero_returns_none() -> None:
# lot_size 0.1 时 0.05 会舍成 0
precisions = {"BTCUSDT": {"lot_size": "0.1", "price_filter": "0.01"}}
params = build_close_order_params(
symbol="BTCUSDT",
position_side="LONG",
position_amt=0.05,
current_price=Decimal("50000"),
precisions=precisions,
)
assert params is None
def test_build_close_order_params_custom_ratios() -> None:
precisions = {"BTCUSDT": {"lot_size": "0.001", "price_filter": "0.01"}}
params = build_close_order_params(
symbol="BTCUSDT",
position_side="LONG",
position_amt=0.1,
current_price=Decimal("50000"),
precisions=precisions,
long_ratio=Decimal("1.01"),
short_ratio=Decimal("0.99"),
)
assert params is not None
assert Decimal(params["price"]) == Decimal("50500") # 50000 * 1.01

216
tests/test_position_closer_flow.py Normal file
View File

@@ -0,0 +1,216 @@
"""
除平仓外的流程单元测试：拉持仓、解析精度、dry_run 不下单。
使用 mock 替代真实 Binance/Redis，验证 get_positions、get_symbol_precision、
place_close_order(dry_run=True) 不调用 futures_create_order。
"""
import sys
from decimal import Decimal
from pathlib import Path
from unittest.mock import AsyncMock
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from position_closer import (
filter_nonzero_positions,
get_min_unrealized_profit_in_window,
get_positions,
get_symbol_precision,
get_total_unrealized_profit,
parse_exchange_info_to_precisions,
place_close_order,
save_unrealized_profit_history,
)
# ----- get_total_unrealized_profit：账户总未实现盈亏 -----
@pytest.mark.asyncio
async def test_get_total_unrealized_profit_sums_all_positions() -> None:
raw_positions = [
{"symbol": "BTCUSDT", "unrealizedProfit": "10.5"},
{"symbol": "ETHUSDT", "unrealizedProfit": "-2"},
{"symbol": "XRPUSDT", "unrealizedProfit": "0.5"},
]
client = AsyncMock()
client.futures_account = AsyncMock(return_value={"positions": raw_positions})
result = await get_total_unrealized_profit(client)
assert result == pytest.approx(10.5 - 2 + 0.5)
assert client.futures_account.await_count == 1
# ----- get_positions：拉取持仓并过滤为零 -----
@pytest.mark.asyncio
async def test_get_positions_returns_only_nonzero() -> None:
raw_positions = [
{"symbol": "BTCUSDT", "positionAmt": "0.1", "notional": "5000"},
{"symbol": "ETHUSDT", "positionAmt": "0", "notional": "0"},
{"symbol": "XRPUSDT", "positionAmt": "-1", "notional": "2000"},
]
client = AsyncMock()
client.futures_account = AsyncMock(return_value={"positions": raw_positions})
result = await get_positions(client)
assert result == filter_nonzero_positions(raw_positions)
assert len(result) == 2
assert client.futures_account.await_count == 1
@pytest.mark.asyncio
async def test_get_positions_empty_account() -> None:
client = AsyncMock()
client.futures_account = AsyncMock(return_value={"positions": []})
result = await get_positions(client)
assert result == []
assert client.futures_account.await_count == 1
# ----- get_symbol_precision：拉取并解析精度 -----
@pytest.mark.asyncio
async def test_get_symbol_precision_parses_exchange_info() -> None:
info = {
"symbols": [
{
"symbol": "BTCUSDT",
"filters": [
{"filterType": "LOT_SIZE", "stepSize": "0.001"},
{"filterType": "PRICE_FILTER", "tickSize": "0.1"},
],
}
]
}
client = AsyncMock()
client.futures_exchange_info = AsyncMock(return_value=info)
result = await get_symbol_precision(client)
assert result == parse_exchange_info_to_precisions(info)
assert result["BTCUSDT"] == {"lot_size": "0.001", "price_filter": "0.1"}
assert client.futures_exchange_info.await_count == 1
# ----- place_close_order(dry_run=True)：不调用下单 API -----
@pytest.mark.asyncio
async def test_place_close_order_dry_run_does_not_call_futures_create_order() -> None:
client = AsyncMock()
client.futures_create_order = AsyncMock()
precisions = {"BTCUSDT": {"lot_size": "0.001", "price_filter": "0.1"}}
await place_close_order(
client,
symbol="BTCUSDT",
position_side="LONG",
position_amt=0.1,
current_price=Decimal("50000"),
precisions=precisions,
reduce_only=True,
dry_run=True,
)
client.futures_create_order.assert_not_called()
@pytest.mark.asyncio
async def test_place_close_order_not_dry_run_calls_futures_create_order() -> None:
client = AsyncMock()
client.futures_create_order = AsyncMock(return_value={"orderId": 123})
precisions = {"BTCUSDT": {"lot_size": "0.001", "price_filter": "0.1"}}
await place_close_order(
client,
symbol="BTCUSDT",
position_side="LONG",
position_amt=0.1,
current_price=Decimal("50000"),
precisions=precisions,
reduce_only=True,
dry_run=False,
)
client.futures_create_order.assert_called_once()
call_kw = client.futures_create_order.call_args[1]
assert call_kw["symbol"] == "BTCUSDT"
assert call_kw["side"] == "SELL"
assert call_kw["quantity"] == "0.1"
assert "price" in call_kw
# ----- save_unrealized_profit_history / get_min_unrealized_profit_in_window -----
@pytest.mark.asyncio
async def test_save_unrealized_profit_history_calls_zadd_and_zremrangebyscore() -> None:
from unittest.mock import MagicMock, patch
mock_redis = MagicMock()
mock_redis.zadd = AsyncMock()
mock_redis.zremrangebyscore = AsyncMock()
mock_redis.aclose = AsyncMock()
with patch("redis.asyncio.Redis") as MockRedis:
MockRedis.from_url.return_value = mock_redis
await save_unrealized_profit_history(
"redis://localhost/0",
"test:history",
12.5,
300,
)
mock_redis.zadd.assert_called_once()
call_args = mock_redis.zadd.call_args[0]
assert call_args[0] == "test:history"
assert isinstance(call_args[1], dict)
member = list(call_args[1].keys())[0]
assert ":12.5" in member
mock_redis.zremrangebyscore.assert_called_once()
mock_redis.aclose.assert_called_once()
@pytest.mark.asyncio
async def test_get_min_unrealized_profit_in_window_returns_min() -> None:
from unittest.mock import MagicMock, patch
mock_redis = MagicMock()
mock_redis.zrangebyscore = AsyncMock(
return_value=["1000.0:5.0", "1001.0:3.0", "1002.0:7.0"]
)
mock_redis.aclose = AsyncMock()
with patch("redis.asyncio.Redis") as MockRedis:
MockRedis.from_url.return_value = mock_redis
result = await get_min_unrealized_profit_in_window(
"redis://localhost/0",
"test:history",
300,
)
assert result == 3.0
mock_redis.zrangebyscore.assert_called_once()
mock_redis.aclose.assert_called_once()
@pytest.mark.asyncio
async def test_get_min_unrealized_profit_in_window_empty_returns_none() -> None:
from unittest.mock import MagicMock, patch
mock_redis = MagicMock()
mock_redis.zrangebyscore = AsyncMock(return_value=[])
mock_redis.aclose = AsyncMock()
with patch("redis.asyncio.Redis") as MockRedis:
MockRedis.from_url.return_value = mock_redis
result = await get_min_unrealized_profit_in_window(
"redis://localhost/0",
"test:history",
300,
)
assert result is None

374
tests/test_position_closer_logic.py Normal file
View File

@@ -0,0 +1,374 @@
"""
position_closer 模块中纯逻辑函数的单元测试。
"""
import sys
from decimal import Decimal
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from position_closer import (
LONG_CLOSE_PRICE_RATIO,
SHORT_CLOSE_PRICE_RATIO,
build_close_order_params,
filter_nonzero_positions,
get_notional_by_side,
get_symbols_to_close,
get_unrealized_profit,
group_positions_by_symbol,
parse_exchange_info_to_precisions,
parse_ticker_message,
round_to_step,
should_close_symbol,
)
# ----- round_to_step -----
@pytest.mark.parametrize(
"value, step, expected",
[
(Decimal("1.234"), "0.01", "1.23"),
(Decimal("1.239"), "0.01", "1.23"),
(Decimal("0.001"), "0.001", "0.001"),
(Decimal("0.0015"), "0.001", "0.001"),
(Decimal("100"), "1", "1E+2"), # normalize() 输出科学计数
(Decimal("1.5"), "0", "1.5"),
(Decimal("1.5"), "0.1", "1.5"),
],
)
def test_round_to_step(value: Decimal, step: str, expected: str) -> None:
assert round_to_step(value, step) == expected
# ----- group_positions_by_symbol -----
def test_group_positions_by_symbol_empty() -> None:
assert group_positions_by_symbol([]) == {}
def test_group_positions_by_symbol_single() -> None:
positions = [{"symbol": "BTCUSDT", "positionAmt": "0.1"}]
got = group_positions_by_symbol(positions)
assert list(got.keys()) == ["BTCUSDT"]
assert got["BTCUSDT"] == positions
def test_group_positions_by_symbol_multiple_symbols() -> None:
positions = [
{"symbol": "BTCUSDT", "positionAmt": "0.1"},
{"symbol": "ETHUSDT", "positionAmt": "1"},
{"symbol": "BTCUSDT", "positionAmt": "0.2", "positionSide": "SHORT"},
]
got = group_positions_by_symbol(positions)
assert set(got.keys()) == {"BTCUSDT", "ETHUSDT"}
assert len(got["BTCUSDT"]) == 2
assert len(got["ETHUSDT"]) == 1
def test_group_positions_by_symbol_skips_empty_symbol() -> None:
positions = [{"symbol": "", "positionAmt": "0.1"}, {"symbol": "BTCUSDT", "positionAmt": "0.1"}]
got = group_positions_by_symbol(positions)
assert list(got.keys()) == ["BTCUSDT"]
# ----- filter_nonzero_positions -----
def test_filter_nonzero_positions_empty() -> None:
assert filter_nonzero_positions([]) == []
def test_filter_nonzero_positions_keeps_nonzero() -> None:
positions = [
{"symbol": "BTCUSDT", "positionAmt": "0.1"},
{"symbol": "ETHUSDT", "positionAmt": "-1"},
]
assert len(filter_nonzero_positions(positions)) == 2
def test_filter_nonzero_positions_removes_zero() -> None:
positions = [
{"symbol": "BTCUSDT", "positionAmt": "0"},
{"symbol": "ETHUSDT", "positionAmt": "0.0"},
{"symbol": "XRPUSDT", "positionAmt": ""},
]
assert filter_nonzero_positions(positions) == []
# ----- get_notional_by_side -----
def test_get_notional_by_side_empty() -> None:
assert get_notional_by_side([]) == (0.0, 0.0)
def test_get_notional_by_side_both_long() -> None:
positions = [{"positionAmt": "0.1", "notional": "5000", "positionSide": "BOTH"}]
assert get_notional_by_side(positions) == (5000.0, 0.0)
def test_get_notional_by_side_both_short() -> None:
positions = [{"positionAmt": "-0.1", "notional": "5000", "positionSide": "BOTH"}]
assert get_notional_by_side(positions) == (0.0, 5000.0)
def test_get_notional_by_side_hedge_mode() -> None:
positions = [
{"positionAmt": "0.1", "notional": "5000", "positionSide": "LONG"},
{"positionAmt": "-0.05", "notional": "2500", "positionSide": "SHORT"},
]
assert get_notional_by_side(positions) == (5000.0, 2500.0)
def test_get_notional_by_side_uses_abs_notional() -> None:
positions = [{"positionAmt": "-0.1", "notional": "-5000", "positionSide": "BOTH"}]
assert get_notional_by_side(positions) == (0.0, 5000.0)
# ----- get_unrealized_profit -----
def test_get_unrealized_profit_empty() -> None:
assert get_unrealized_profit([]) == 0.0
# ----- should_close_symbol -----
@pytest.mark.parametrize(
"long_n, short_n, unrealized_profit, threshold, in_redis, small_threshold, small_min_profit, large_min_profit, expected",
[
(0, 0, 0, 50, False, 30, 0.03, 0.3, False),
(60, 0, 0.5, 50, False, 30, 0.03, 0.3, True), # 大仓位且盈利 > 0.3
(60, 0, 0.2, 50, False, 30, 0.03, 0.3, False),
(60, 0, 0, 50, False, 30, 0.03, 0.3, False),
(40, 0, 0.5, 50, False, 30, 0.03, 0.3, False),
(20, 0, 0.5, 50, False, 30, 0.03, 0.3, True),
(20, 0, 0.05, 50, False, 30, 0.03, 0.3, True),
(20, 0, 0.02, 50, False, 30, 0.03, 0.3, False),
(20, 0, 0, 50, False, 30, 0.03, 0.3, False),
(0, 0, 0, 50, True, 30, 0.03, 0.3, True),
(30, 30, 0.5, 50, False, 30, 0.03, 0.3, True), # 大仓位且盈利>0.3
],
)
def test_should_close_symbol(
long_n: float,
short_n: float,
unrealized_profit: float,
threshold: float,
in_redis: bool,
small_threshold: float,
small_min_profit: float,
large_min_profit: float,
expected: bool,
) -> None:
assert (
should_close_symbol(
long_n, short_n, unrealized_profit, threshold, in_redis,
small_threshold, small_min_profit, large_min_profit,
)
is expected
)
# ----- get_symbols_to_close -----
def test_get_symbols_to_close_empty() -> None:
symbols, filtered = get_symbols_to_close({}, 50, set())
assert symbols == set()
assert filtered == {}
def test_get_symbols_to_close_by_threshold() -> None:
# 大仓位且盈利>0.3 触发
by_symbol = {
"BTCUSDT": [
{"positionAmt": "0.1", "notional": "15000", "positionSide": "BOTH", "unrealizedProfit": "10"},
],
"ETHUSDT": [
{"positionAmt": "1", "notional": "3000", "positionSide": "BOTH", "unrealizedProfit": "0"},
],
}
symbols, filtered = get_symbols_to_close(by_symbol, 50, set(), 30, 0.03, 0.3)
assert symbols == {"BTCUSDT"}
assert set(filtered.keys()) == {"BTCUSDT"}
def test_get_symbols_to_close_by_redis() -> None:
by_symbol = {
"ETHUSDT": [
{"positionAmt": "1", "notional": "3000", "positionSide": "BOTH", "unrealizedProfit": "0"},
],
}
symbols, filtered = get_symbols_to_close(by_symbol, 50, {"ETHUSDT"})
assert symbols == {"ETHUSDT"}
assert filtered["ETHUSDT"] == by_symbol["ETHUSDT"]
def test_get_symbols_to_close_small_and_profit() -> None:
by_symbol = {
"XRPUSDT": [
{"positionAmt": "10", "notional": "20", "positionSide": "BOTH", "unrealizedProfit": "0.5"},
],
}
symbols, filtered = get_symbols_to_close(by_symbol, 50, set(), small_threshold=30, small_min_profit=0.03)
assert symbols == {"XRPUSDT"}
assert list(filtered.keys()) == ["XRPUSDT"]
def test_get_symbols_to_close_small_profit_below_min() -> None:
by_symbol = {
"XRPUSDT": [
{"positionAmt": "10", "notional": "20", "positionSide": "BOTH", "unrealizedProfit": "0.02"},
],
}
symbols, filtered = get_symbols_to_close(by_symbol, 50, set(), small_threshold=30, small_min_profit=0.03)
assert symbols == set()
assert filtered == {}
# ----- parse_ticker_message -----
def test_parse_ticker_message_valid() -> None:
msg = {"stream": "btcusdt@ticker", "data": {"c": "50000.5", "s": "BTCUSDT"}}
symbol, price = parse_ticker_message(msg)
assert symbol == "BTCUSDT"
assert price == Decimal("50000.5")
def test_parse_ticker_message_no_stream() -> None:
msg = {"data": {"c": "50000"}}
symbol, price = parse_ticker_message(msg)
assert symbol is None
assert price is None
def test_parse_ticker_message_no_price() -> None:
msg = {"stream": "btcusdt@ticker", "data": {"s": "BTCUSDT"}}
symbol, price = parse_ticker_message(msg)
assert symbol == "BTCUSDT"
assert price is None
def test_parse_ticker_message_invalid_price() -> None:
msg = {"stream": "btcusdt@ticker", "data": {"c": "not-a-number"}}
symbol, price = parse_ticker_message(msg)
assert symbol == "BTCUSDT"
assert price is None
# ----- parse_exchange_info_to_precisions -----
def test_parse_exchange_info_to_precisions_empty() -> None:
assert parse_exchange_info_to_precisions({}) == {}
assert parse_exchange_info_to_precisions({"symbols": []}) == {}
def test_parse_exchange_info_to_precisions_defaults() -> None:
info = {"symbols": [{"symbol": "BTCUSDT", "filters": []}]}
got = parse_exchange_info_to_precisions(info)
assert got["BTCUSDT"] == {"lot_size": "0.01", "price_filter": "0.01"}
def test_parse_exchange_info_to_precisions_with_filters() -> None:
info = {
"symbols": [
{
"symbol": "BTCUSDT",
"filters": [
{"filterType": "LOT_SIZE", "stepSize": "0.001"},
{"filterType": "PRICE_FILTER", "tickSize": "0.1"},
],
}
]
}
got = parse_exchange_info_to_precisions(info)
assert got["BTCUSDT"] == {"lot_size": "0.001", "price_filter": "0.1"}
def test_parse_exchange_info_skips_empty_symbol() -> None:
info = {"symbols": [{"symbol": "", "filters": []}]}
assert parse_exchange_info_to_precisions(info) == {}
# ----- build_close_order_params -----
def test_build_close_order_params_long() -> None:
precisions = {"BTCUSDT": {"lot_size": "0.001", "price_filter": "0.1"}}
params = build_close_order_params(
symbol="BTCUSDT",
position_side="LONG",
position_amt=0.1,
current_price=Decimal("50000"),
precisions=precisions,
reduce_only=True,
)
assert params is not None
assert params["symbol"] == "BTCUSDT"
assert params["side"] == "SELL"
assert params["positionSide"] == "LONG"
assert params["quantity"] == "0.1"
assert Decimal(params["price"]) == Decimal("50150") # 50000 * 1.003
assert params.get("reduceOnly") == "true"
def test_build_close_order_params_short() -> None:
precisions = {"BTCUSDT": {"lot_size": "0.001", "price_filter": "0.1"}}
params = build_close_order_params(
symbol="BTCUSDT",
position_side="SHORT",
position_amt=-0.1,
current_price=Decimal("50000"),
precisions=precisions,
reduce_only=True,
)
assert params is not None
assert params["side"] == "BUY"
assert params["positionSide"] == "SHORT"
assert Decimal(params["price"]) == Decimal("49850") # 50000 * 0.997
def test_build_close_order_params_both_long() -> None:
precisions = {"BTCUSDT": {"lot_size": "0.001", "price_filter": "0.01"}}
params = build_close_order_params(
symbol="BTCUSDT",
position_side="BOTH",
position_amt=0.05,
current_price=Decimal("50000"),
precisions=precisions,
reduce_only=True,
)
assert params is not None
assert params["side"] == "SELL"
assert params["quantity"] == "0.05"
assert "positionSide" not in params or params.get("positionSide") == "BOTH"
assert params.get("reduceOnly") == "true"
def test_build_close_order_params_zero_amt_returns_none() -> None:
params = build_close_order_params(
symbol="BTCUSDT",
position_side="LONG",
position_amt=0,
current_price=Decimal("50000"),
precisions={},
)
assert params is None
def test_build_close_order_params_tiny_amt_rounds_to_zero_returns_none() -> None:
# lot_size 0.1 时 0.05 会舍成 0
precisions = {"BTCUSDT": {"lot_size": "0.1", "price_filter": "0.01"}}
params = build_close_order_params(
symbol="BTCUSDT",
position_side="LONG",
position_amt=0.05,
current_price=Decimal("50000"),
precisions=precisions,
)
assert params is None
def test_build_close_order_params_custom_ratios() -> None:
precisions = {"BTCUSDT": {"lot_size": "0.001", "price_filter": "0.01"}}
params = build_close_order_params(
symbol="BTCUSDT",
position_side="LONG",
position_amt=0.1,
current_price=Decimal("50000"),
precisions=precisions,
long_ratio=Decimal("1.01"),
short_ratio=Decimal("0.99"),
)
assert params is not None
assert Decimal(params["price"]) == Decimal("50500") # 50000 * 1.01