@clawhub-zadanthony-47e69c711d
跨平台资源搜索编排器。搜索 skill、MCP 服务器、提示词模板、开源项目。 覆盖 skills.sh、ClawHub、SkillHub、AI Skills Show、MCPServers.org、 prompts.chat、GitHub 等 14+ 个聚合站。 触发场景：用户说"找个xxx工具"、"有没有xx...
---
name: find-everything
description: >
跨平台资源搜索编排器。搜索 skill、MCP 服务器、提示词模板、开源项目。
覆盖 skills.sh、ClawHub、SkillHub、AI Skills Show、MCPServers.org、
prompts.chat、GitHub 等 14+ 个聚合站。
触发场景：用户说"找个xxx工具"、"有没有xxx skill"、"帮我搜xxx MCP"、
"找提示词"、"有什么好用的xxx"，或显式调用 /find-everything。
也会在检测到用户持续做某类任务且缺少相关工具时主动推荐（每会话最多 1 次）。
---
# find-everything：跨平台资源搜索
## 触发条件
1. **显式调用**：`/find-everything {query}`
2. **自动检测**：用户问"有没有xxx工具"、"帮我找个xxx"、"有没有skill能..."
3. **主动推荐**：基于对话上下文判断用户在持续做某类任务且缺少相关工具（best-effort 启发式，同会话最多 1 次）
## 执行流程
### Step 1: 意图分类
将用户查询分类为一个或多个类别：
- **skill**: Agent Skills（可安装技能）
- **mcp**: MCP Servers
- **prompt**: 提示词模板/角色扮演/图片生成
- **repo**: GitHub 开源项目
模糊需求同时搜索多个类别。从自然语言中提取核心搜索关键词。
### Step 2: 读取注册表并路由
读取 `references/registry.json`。若读取失败，使用以下硬编码最小源：
- skills-sh: `npx skills find {query}`
- github: `gh search repos {query} --sort stars --limit 10 --json name,owner,description,url,stargazersCount`
筛选规则：
1. 只选 `enabled: true` 的源
2. 只选 `category` 包含目标类别的源
3. 检查 `requires` 是否满足：
- `requires` 为 CLI 名称（如 `npx`、`gh`）→ 用 `which` 检查
- `requires` 以 `mcp:` 开头（如 `mcp:prompts-chat`）→ 检查对应 MCP tool 是否在当前会话可用（尝试 ToolSearch 或直接调用）
4. 不可用的源跳过，记录到 `skipped_sources` 列表，附带 `install_hint`（如有）
### Step 3: Tier 1 搜索
在**同一条回复**中发起多个独立 tool 调用实现并行：
**cli 类型**：用 Bash tool 执行 `command` 字段（{query} 替换为实际关键词），15 秒超时。
**mcp 类型**：调用 registry.json 中 `tool` 字段指定的 MCP tool，参数取 `tool_params`（{query} 替换为实际关键词）。例如：
- `search_prompts({ query: "blockchain analyst", limit: 10 })` → 返回 prompt 列表
- `search_skills({ query: "search aggregator", limit: 10 })` → 返回 skill 列表
- 支持的额外过滤参数（按需使用）：`type`（TEXT/STRUCTURED/IMAGE）、`category`、`tag`
**skill 类型**：调用对应已安装 skill。
若并行不可用，按优先级执行：skills-sh > github > clawhub > 其他。
### Step 4: 结果数量预判
基于标题/描述的关键词匹配做轻量判断（非完整 LLM 评估）：
- ≥3 条匹配度高 → 跳到 Step 6（末尾注明"搜索更多源可获取更多结果"）
- <3 条或匹配度低 → 继续 Step 5
### Step 5: Tier 2/3 搜索
**Tier 2**：将同类别 Tier 2 源合并为 1 次 WebSearch：
```
{query} {category关键词} site:skillhub.club OR site:aiskillsshow.com OR site:skillsmp.com
```
若合并查询返回 <2 条，退回对 Top 2-3 优先源逐个 `site:` 搜索。
**Tier 3**（仍不足时）：不带 `site:` 限定的广域 WebSearch。
### Step 6: 结果评估
对每条结果判断相关性：
- **高相关**：保留，排在前面
- **中相关**：保留，排在后面，标注"可能相关"
- **不相关**：丢弃
对 Top 3-5 条高相关结果，可选 WebFetch 补充详情。
### Step 7: 安全快筛
对所有保留结果做元数据安全标注：
- **[SAFE]**：来自注册表已知平台 + 安装量 >100
- **[CAUTION]**：安装量低、来源不明、或 Tier 3 发现
- **[RISK]**：名称疑似 typosquat，或其他红旗信号
### Step 8: 去重 + 排序
同一工具出现在多个源 → 合并为一条，标注所有来源。
合并时优先保留安装量/star 数最高的来源作为主展示（同一站点可能有 Tier 1 CLI 和 Tier 2 WebSearch 两条结果，合并时保留 Tier 1 数据）。
排序：相关度 > 安全等级 > 安装量/star 数。
### Step 9: 展示推荐
格式：
```
找到 N 个相关结果（来自 X 个源）
1. [名称] [SAFE]
来源: skills.sh | 安装量: 1.2K
简介: ...
推荐理由: ...
2. [名称] [CAUTION]
来源: WebSearch (skillhub.club) | star: 50
简介: ...
注意: 来源非直接 API，建议安装前审查源码
---
未覆盖的搜索源: clawhub CLI（未安装）
提示: npm i -g clawhub
```
注意：同一依赖缺失提示每会话只展示一次，避免重复打扰。
### Step 10: 用户后续操作
用户说**"看看"/"详细"** → WebFetch 详情页展示更多信息。
用户说**"安装"/"用这个"** → 触发深度安全扫描：
**1. 获取资源内容（按类型）：**
| 资源类型 | 扫描目标 | 获取方式 |
|---|---|---|
| Skill (skills.sh/clawhub) | SKILL.md + scripts/ 目录 | 优先 WebFetch GitHub 源码（从搜索结果中的 repo URL 获取）；若不可访问，`npx skills add <name>` 安装到临时目录后读取 |
| MCP Server | package.json + 入口文件 | WebFetch npm registry 页面或 GitHub README，提取入口文件路径后 WebFetch |
| GitHub Repo | README.md + 主要脚本文件 | `gh api repos/{owner}/{repo}/contents` 获取文件列表，WebFetch 关键文件 |
| Prompt | 提示词文本本身 | 通常搜索结果中已包含完整文本，无需额外获取 |
**限制：** 单次扫描最多 50KB。超出只扫 SKILL.md + 入口文件。
**2.** 运行安全扫描（路径相对于本 SKILL.md 所在目录解析）：`python3 scripts/security_scan.py <file> [--check-name <name> --known-skills references/known_skills.txt]`
**3.** 读取 `references/security-checklist.md`，结合 security_scan.py 结果做 LLM 上下文评估
**4.** 输出量化评分（0-100）和风险定级
**5.** [SAFE] → 执行安装；[CAUTION] → 展示详情让用户确认；[RISK] → 明确提示风险，不自动安装
### Step 11: 新站点发现
Tier 3 搜索发现了不在 registry.json 中的优质资源站 → 提示用户：
"发现新站点 xxx.com，内容相关度高。要加入搜索注册表吗？"
用户确认 → 在 registry.json 的 sources 数组中追加新条目。
## 错误处理
| 场景 | 处理 |
|---|---|
| CLI 超时（15s） | 跳过该源，继续其他 |
| WebSearch 错误/限流 | 跳过 Tier 2，展示 Tier 1 结果 + 提示稍后重试 |
| security_scan.py 崩溃 | 退回 LLM-only 评估，标记"自动扫描未完成" |
| 所有 Tier 1 不可用 | 直接进 Tier 2，注明"主要搜索源暂不可用" |
| 所有源失败 | 告知用户搜索失败，建议检查网络 |
## 主动推荐逻辑
触发条件（低频，同会话最多 1 次）：
- 对话上下文显示用户持续在做某类任务（如调试、前端开发、数据处理）
- 且当前未见明显相关的 skill/MCP 在使用
推荐格式：
"[发现] 你正在做 xxx，有一些工具可能有帮助，要搜索看看吗？"
FILE:references/known_skills.txt
# 知名 skill 名称列表（用于 typosquat 检测）
# 每行一个名称，# 开头为注释
# 来源：skills.sh 热门搜索结果 + 常见 MCP 服务器
# 更新日期：2026-03-19
# --- 高安装量 skills (skills.sh) ---
ai-rag-pipeline
apify-actor-development
apify-ecommerce
apify-lead-generation
apify-market-research
apify-trend-analysis
apify-ultimate-scraper
apollo-client
apollo-mcp-server
apollo-server
authentication-setup
aws-serverless
aws-solution-architect
azure-deploy
azure-deployment-preflight
azure-storage
baoyu-danger-x-to-markdown
baoyu-format-markdown
baoyu-markdown-to-html
baoyu-url-to-markdown
better-auth-best-practices
better-auth-security-best-practices
clerk-nextjs-patterns
clip-aware-embeddings
cloudflare-vectorize
cloudformation-to-pulumi
contrast-checker
convex-file-storage
create-auth-skill
create-github-action-workflow-specification
create-specification
database-migration
database-schema-design
database-schema-designer
dataverse-python-production-code
dbt-transformation-patterns
debugger
debugging
debugging-strategies
deep-agents-memory
deploy-to-vercel
deployment-automation
doc-coauthoring
docker
docker-best-practices
docker-compose-orchestration
docker-containerization
docker-expert
documentation-writer
email-and-password-best-practices
embeddings
eslint-prettier-config
excalidraw-diagram-generator
expo-cicd-workflows
expo-deployment
expo-tailwind-setup
figma
figma-implement-design
find-skills
firebase-ai-logic
firebase-auth-basics
firebase-basics
firebase-firestore-basics
firebase-hosting-basics
fixing-motion-performance
flutter-working-with-databases
gh-cli
git-commit
go-linting
golang
golang-backend-development
golang-grpc
golang-patterns
golang-pro
golang-testing
google-gemini-embeddings
graphql-operations
graphql-schema
grepai-embeddings-lmstudio
grepai-embeddings-ollama
grepai-embeddings-openai
hindsight-docs
javascript-typescript-jest
json-render-react
k8s-security-policies
kubernetes
kubernetes-architect
kubernetes-deployment
kubernetes-specialist
langchain-fundamentals
langchain-rag
langchain4j-vector-stores-configuration
langgraph-docs
langgraph-fundamentals
langgraph-persistence
legacy-circuit-mockups
linear
lint-and-validate
linting-neostandard-eslint9
markdown-to-html
mcp-deploy-manage-agents
migrate-oxlint
multi-stage-dockerfile
neon-postgres
nextjs-app-router-fundamentals
nextjs-app-router-patterns
nextjs-best-practices
nextjs-developer
nextjs-supabase-auth
obsidian-markdown
openapi-to-typescript
organization-best-practices
organizational-transformation
parallel-debugging
performance
performance-optimization
plantuml-ascii
playwright
playwright-generate-test
power-bi-performance-troubleshooting
prd
prisma-cli
prisma-client-api
prisma-database-setup
prisma-expert
prisma-postgres
prisma-upgrade-v7
pytest-coverage
python-executor
python-mcp-server-generator
python-performance-optimization
python-sdk
python-testing-patterns
qdrant-vector-search
rag-implementation
react
react-email
refactor
resume
rust-async-patterns
rust-best-practices
rust-engineer
rust-mcp-server-generator
rust-refactor-helper
rust-router
screenshot
security-best-practices
security-requirement-extraction
security-review
seo-geo
skill-creator
skill-lookup
skill-vetter
prompt-lookup
clawhub
deep-research
exa-web-search-free
solidity-security
spreadsheet
supabase
supabase-best-practices
supabase-nextjs
supabase-postgres-best-practices
swift-concurrency-pro
swift-testing-pro
swiftdata-pro
swiftui-performance-audit
swiftui-pro
systematic-debugging
tailwind-css-patterns
tailwind-design-system
tailwind-patterns
tailwind-v4-shadcn
tailwindcss
tailwindcss-advanced-layouts
tailwindcss-animations
tailwindcss-mobile-first
terraform-azurerm-set-diff-analyzer
terraform-engineer
terraform-module-library
terraform-stacks
terraform-style-guide
terraform-test
test-driven-development
typescript-advanced-types
typescript-expert
typescript-mcp-server-generator
typescript-react-reviewer
unocss
vector-index-tuning
vectorbt-expert
vercel-react-best-practices
vercel-react-native-skills
vue
vue-best-practices
vue-debug-guides
vue-pinia-best-practices
vue-router-best-practices
vueuse-functions
webapp-testing
# --- 常见 MCP 服务器 ---
playwright
puppeteer
filesystem
github
slack
postgres
sqlite
redis
brave-search
memory
sequential-thinking
fetch
everart
google-maps
sentry
FILE:references/registry.json
{
"version": "1.1",
"categories": {
"skill": "Agent Skills（可安装技能）",
"mcp": "MCP Servers",
"prompt": "提示词模板/角色扮演/图片生成",
"repo": "GitHub 开源项目"
},
"sources": [
{
"id": "skills-sh",
"name": "Skills.sh",
"url": "https://skills.sh",
"category": ["skill"],
"tier": 1,
"method": "cli",
"command": "npx skills find {query}",
"requires": "npx",
"enabled": true
},
{
"id": "github",
"name": "GitHub",
"url": "https://github.com",
"category": ["repo", "mcp", "skill"],
"tier": 1,
"method": "cli",
"command": "gh search repos {query} --sort stars --limit 10 --json name,owner,description,url,stargazersCount",
"requires": "gh",
"enabled": true
},
{
"id": "clawhub-cli",
"name": "ClawHub (CLI)",
"url": "https://clawhub.ai",
"category": ["skill"],
"tier": 1,
"method": "cli",
"command": "clawhub search {query}",
"requires": "clawhub",
"install_hint": "npm i -g clawhub",
"enabled": true
},
{
"id": "prompts-chat-prompts",
"name": "Prompts.chat (提示词 MCP)",
"url": "https://prompts.chat",
"category": ["prompt"],
"tier": 1,
"method": "mcp",
"tool": "search_prompts",
"tool_params": {
"query": "{query}",
"limit": 10
},
"requires": "mcp:prompts-chat",
"mcp_package": "@fkadev/prompts.chat-mcp",
"install_hint": "claude mcp add prompts-chat -- npx -y @fkadev/prompts.chat-mcp",
"enabled": true,
"verified": "2026-03-19"
},
{
"id": "prompts-chat-skills",
"name": "Prompts.chat (Skills MCP)",
"url": "https://prompts.chat",
"category": ["skill"],
"tier": 1,
"method": "mcp",
"tool": "search_skills",
"tool_params": {
"query": "{query}",
"limit": 10
},
"requires": "mcp:prompts-chat",
"mcp_package": "@fkadev/prompts.chat-mcp",
"install_hint": "claude mcp add prompts-chat -- npx -y @fkadev/prompts.chat-mcp",
"enabled": true,
"verified": "2026-03-19"
},
{
"id": "clawhub-web",
"name": "ClawHub",
"url": "https://clawhub.ai",
"category": ["skill"],
"tier": 2,
"method": "web_search",
"search_prefix": "site:clawhub.ai",
"enabled": true
},
{
"id": "prompts-chat-web",
"name": "Prompts.chat",
"url": "https://prompts.chat",
"category": ["prompt", "skill"],
"tier": 2,
"method": "web_search",
"search_prefix": "site:prompts.chat",
"enabled": true
},
{
"id": "skillhub-club",
"name": "SkillHub",
"url": "https://www.skillhub.club",
"category": ["skill"],
"tier": 2,
"method": "web_search",
"search_prefix": "site:skillhub.club",
"enabled": true
},
{
"id": "mcpservers",
"name": "MCPServers.org",
"url": "https://mcpservers.org",
"category": ["mcp", "skill"],
"tier": 2,
"method": "web_search",
"search_prefix": "site:mcpservers.org",
"enabled": true
},
{
"id": "skillsmp",
"name": "SkillsMP",
"url": "https://skillsmp.com",
"category": ["skill"],
"tier": 2,
"method": "web_search",
"search_prefix": "site:skillsmp.com",
"note": "有 API (skillsmp.com/docs/api)，后续可升级为 Tier 1",
"enabled": true
},
{
"id": "aishort",
"name": "AI Short",
"url": "https://www.aishort.top",
"category": ["prompt"],
"tier": 2,
"method": "web_search",
"search_prefix": "site:aishort.top",
"enabled": true
},
{
"id": "nanoprompts",
"name": "NanoPrompts",
"url": "https://nanoprompts.org",
"category": ["prompt"],
"tier": 2,
"method": "web_search",
"search_prefix": "site:nanoprompts.org",
"enabled": true
},
{
"id": "aiart-pics",
"name": "AI Art Pics",
"url": "https://aiart.pics",
"category": ["prompt"],
"tier": 2,
"method": "web_search",
"search_prefix": "site:aiart.pics",
"enabled": true
},
{
"id": "localbanana",
"name": "LocalBanana",
"url": "https://www.localbanana.io",
"category": ["prompt"],
"tier": 2,
"method": "web_search",
"search_prefix": "site:localbanana.io",
"enabled": true
},
{
"id": "aiskillsshow",
"name": "AI Skills Show",
"url": "https://aiskillsshow.com",
"category": ["skill"],
"tier": 2,
"method": "web_search",
"search_prefix": "site:aiskillsshow.com",
"enabled": false,
"disabled_reason": "搜索引擎未收录，WebSearch 返回零结果（2026-03-19 测试）"
}
]
}
FILE:references/security-checklist.md
# 安全评估指南
供 LLM 在深度安全扫描时参考。配合 security_scan.py 的确定性检查结果使用。
## 1. 误报过滤
security_scan.py 可能产生误报。以下情况应降级或忽略：
| 检测结果 | 误报场景 | 处理 |
|---|---|---|
| "ignore previous instructions" | 出现在安全文档/教学材料中 | 降级为 info |
| curl/wget | 下载公开资源（如 GitHub release） | 检查目标 URL 是否可信 |
| eval()/exec() | 模板引擎或 REPL 工具的正常用法 | 看上下文，检查输入是否用户可控 |
| sudo | 安装系统依赖的文档说明 | 区分"说明文档"和"自动执行" |
| .env 访问 | 工具本身需要读取配置 | 检查是否只读且不外传 |
| permission_scope: network | 文档中提到 URL（如"Visit https://..."）但不实际调用 | 检查是否有 fetch()/curl 等实际调用代码，纯文本 URL 忽略 |
## 2. 量化评分体系（0-100）
### 来源可信度（0-25 分）
- 25: 来自 skills.sh/clawhub 等已知平台，安装量 >1000
- 20: 已知平台，安装量 100-1000
- 15: 已知平台，安装量 <100
- 10: GitHub 开源但非聚合平台
- 5: WebSearch 发现，来源不明
- 0: 无法确认来源
### 代码透明度（0-25 分）
- 25: GitHub 开源，代码清晰可审查，无混淆
- 20: 开源但部分代码复杂
- 15: 开源但包含压缩/打包代码
- 10: 部分开源
- 5: 仅有 SKILL.md，无源码
- 0: 存在混淆或加密内容
### 权限范围（0-20 分）
- 20: 仅 fileRead 或无特殊权限
- 15: fileRead + fileWrite
- 10: 需要 network 但有合理理由
- 5: 需要 shell 但有合理理由
- 0: network + shell 组合（数据外传风险）
### 网络风险（0-15 分）
- 15: 无外部网络调用
- 10: 仅调用已知可信 API
- 5: 调用外部 API 但有文档说明
- 0: 调用未知/可疑外部地址
### 社区信号（0-15 分）
- 15: >100 star，活跃维护（最近 30 天有更新），作者有其他知名项目
- 10: 10-100 star，定期维护
- 5: <10 star 但有合理更新历史
- 0: 无 star，无维护历史，或作者账号异常
## 3. 风险定级
| 评分 | 等级 | 建议 |
|---|---|---|
| 80-100 | [SAFE] 安全 | 可以安装/使用 |
| 60-79 | [CAUTION] 谨慎 | 建议审查源码后使用 |
| <60 | [RISK] 风险 | 不推荐安装，存在安全隐患 |
## 4. 评估输出格式
```
安全检查结果 [名称]:
- 安全评分: XX/100
- 来源可信度: XX/25（具体说明）
- 代码透明度: XX/25（具体说明）
- 权限范围: XX/20（具体说明）
- 网络风险: XX/15（具体说明）
- 社区信号: XX/15（具体说明）
- security_scan.py 检测: X 个发现（Y critical, Z high）
- 风险等级: [SAFE/CAUTION/RISK]
- 建议: ...
```
FILE:scripts/security_scan.py
#!/usr/bin/env python3
"""
security_scan.py — 确定性安全检查脚本
用于扫描 skill/prompt/MCP 文件中的安全隐患
用法:
python3 security_scan.py <file_path> [--check-name <name>] [--known-skills <path>]
python3 security_scan.py --version
输出: JSON 到 stdout
退出码: 0=clean, 2=有 critical/high 发现, 1=运行错误
"""
import argparse
import base64
import json
import re
import sys
import unicodedata
from pathlib import Path
__version__ = "1.1.0"
# === Unicode 同形字归一化 ===
# 常见西里尔/希腊字母 → ASCII 映射
CONFUSABLES = {
"\u0430": "a", "\u0435": "e", "\u043e": "o", "\u0440": "p",
"\u0441": "c", "\u0445": "x", "\u0443": "y", "\u0456": "i",
"\u0422": "T", "\u041d": "H", "\u041c": "M", "\u0412": "B",
"\u0410": "A", "\u0415": "E", "\u041e": "O", "\u0421": "C",
"\u03bf": "o", "\u03b1": "a", "\u03b5": "e", # 希腊
"\uff41": "a", "\uff42": "b", "\uff43": "c", # 全角
}
def normalize_unicode(text):
"""NFKC 归一化 + 替换常见同形字"""
text = unicodedata.normalize("NFKC", text)
for char, replacement in CONFUSABLES.items():
text = text.replace(char, replacement)
return text
# === 检测规则 ===
# Prompt injection 模式（severity: critical/high/medium）
INJECTION_PATTERNS = [
# Critical
{"pattern": r"ignore\s+(all\s+)?previous\s+instructions", "severity": "critical", "name": "ignore_instructions"},
{"pattern": r"forget\s+(everything|all)\s+(above|before)", "severity": "critical", "name": "forget_instructions"},
{"pattern": r"you\s+are\s+now\b", "severity": "critical", "name": "role_override"},
{"pattern": r"system\s+prompt\s+override", "severity": "critical", "name": "system_override"},
{"pattern": r"new\s+instructions?\s*:", "severity": "critical", "name": "new_instructions"},
# High
{"pattern": r"\[(SYSTEM|ADMIN|ROOT|ASSISTANT)\]", "severity": "high", "name": "fake_role_tag"},
{"pattern": r"debug\s+mode\s*:\s*(enabled|on|true)", "severity": "high", "name": "debug_mode"},
{"pattern": r"safety\s+mode\s*:\s*(off|disabled|false)", "severity": "high", "name": "safety_mode_off"},
{"pattern": r"admin\s+mode\s*(activated|enabled|on)", "severity": "high", "name": "admin_mode"},
# Medium
{"pattern": r"note\s+to\s+(ai|assistant|agent)\s*:", "severity": "medium", "name": "note_to_ai"},
{"pattern": r"(?:i[''`]?m|i\s+am)\s+the\s+developer", "severity": "medium", "name": "social_developer"},
{"pattern": r"the\s+security\s+check\s+is\s+broken", "severity": "medium", "name": "social_bypass"},
{"pattern": r"other\s+ai\s+assistants?\s+do\s+this", "severity": "medium", "name": "social_pressure"},
]
# 危险命令模式（2.1 修复：curl/wget 模式放宽，支持中间有 flags）
DANGEROUS_PATTERNS = [
{"pattern": r"curl\s+.*\|\s*(?:ba)?sh", "severity": "critical", "name": "curl_pipe_bash"},
{"pattern": r"wget\s+.*\|\s*(?:ba)?sh", "severity": "critical", "name": "wget_pipe_bash"},
{"pattern": r"\brm\s+-[rR]f\b", "severity": "high", "name": "rm_rf"},
{"pattern": r"\bsudo\b", "severity": "high", "name": "sudo"},
{"pattern": r"\bchmod\s+777\b", "severity": "high", "name": "chmod_777"},
{"pattern": r"\beval\s*\(", "severity": "medium", "name": "eval_call"},
{"pattern": r"\bexec\s*\(", "severity": "medium", "name": "exec_call"},
{"pattern": r"\bchild_process\b", "severity": "medium", "name": "child_process"},
{"pattern": r"\bspawn\s*\(", "severity": "medium", "name": "spawn_call"},
{"pattern": r"dangerouslyDisableSandbox", "severity": "critical", "name": "disable_sandbox"},
{"pattern": r"--no-verify", "severity": "medium", "name": "no_verify"},
]
# 数据外传模式
DATA_EXFIL_PATTERNS = [
{"pattern": r"curl\s+.*(?:-d\s+@|--data.*@|--upload-file)", "severity": "critical", "name": "curl_data_exfil"},
{"pattern": r"wget\s+.*--post-file", "severity": "critical", "name": "wget_data_exfil"},
{"pattern": r"\bnc\s+\S+\s+\d+", "severity": "high", "name": "netcat_connection"},
{"pattern": r"curl\s+.*-X\s+POST", "severity": "medium", "name": "curl_post"},
]
# 凭证访问模式（3.3 修复：增加 $HOME 变体）
CREDENTIAL_PATTERNS = [
{"pattern": r"~/\.ssh\b|\.ssh/|\$\{?HOME\}?/\.ssh", "severity": "high", "name": "ssh_access"},
{"pattern": r"~/\.aws\b|\.aws/|\$\{?HOME\}?/\.aws", "severity": "high", "name": "aws_access"},
{"pattern": r"~/\.env\b|\.env\b|\$\{?HOME\}?/\.env", "severity": "high", "name": "env_access"},
{"pattern": r"\.credentials\b|credentials\.json", "severity": "high", "name": "credentials_access"},
{"pattern": r"~/\.bashrc|~/\.zshrc|~/\.profile", "severity": "medium", "name": "shell_config_access"},
{"pattern": r"\bcrontab\b|/etc/cron", "severity": "medium", "name": "crontab_access"},
]
# CSS/样式隐藏
OBFUSCATION_PATTERNS = [
{"pattern": r"display\s*:\s*none", "severity": "medium", "name": "css_display_none"},
{"pattern": r"visibility\s*:\s*hidden", "severity": "medium", "name": "css_visibility_hidden"},
{"pattern": r"font-size\s*:\s*0", "severity": "medium", "name": "css_font_zero"},
]
# 零宽字符
ZERO_WIDTH_CHARS = ["\u200b", "\u200c", "\u200d", "\ufeff"]
# HTML 注释中的隐藏指令
HTML_COMMENT_RE = re.compile(r"<!--(.*?)-->", re.DOTALL)
MAX_SCAN_SIZE = 50 * 1024 # 50KB
def decode_base64_strings(text):
"""查找并解码 base64 字符串（支持递归一层解码）"""
findings = []
b64_pattern = re.compile(r"[A-Za-z0-9+/]{20,}={0,2}")
def _check_decoded(decoded, original_match, line_offset):
"""检查解码内容是否可疑"""
if any(re.search(p["pattern"], decoded, re.IGNORECASE) for p in INJECTION_PATTERNS):
findings.append({
"category": "obfuscation",
"severity": "high",
"pattern": "base64_injection",
"location": {"line": line_offset, "context": original_match[:60]},
"raw_match": f"base64 decoded: {decoded[:100]}",
})
return True
return False
for match in b64_pattern.finditer(text):
try:
decoded = base64.b64decode(match.group()).decode("utf-8", errors="ignore")
line_num = text[:match.start()].count("\n") + 1
if _check_decoded(decoded, match.group(), line_num):
continue
# 3.1 修复：递归一层，检查双重 base64
inner_matches = b64_pattern.findall(decoded)
for inner in inner_matches:
try:
inner_decoded = base64.b64decode(inner).decode("utf-8", errors="ignore")
_check_decoded(inner_decoded, match.group(), line_num)
except Exception:
pass
except Exception:
pass
return findings
def detect_zero_width(text):
"""检测零宽字符"""
findings = []
for i, line in enumerate(text.split("\n"), 1):
for char in ZERO_WIDTH_CHARS:
if char in line:
findings.append({
"category": "obfuscation",
"severity": "medium",
"pattern": "zero_width_char",
"location": {"line": i, "context": repr(line[:80])},
"raw_match": f"U+{ord(char):04X}",
})
break # 每行只报一次
return findings
def detect_html_hidden(text):
"""检测 HTML 注释中的隐藏指令"""
findings = []
for match in HTML_COMMENT_RE.finditer(text):
content = match.group(1).strip()
for p in INJECTION_PATTERNS + DANGEROUS_PATTERNS:
if re.search(p["pattern"], content, re.IGNORECASE):
findings.append({
"category": "prompt_injection",
"severity": "high",
"pattern": "html_comment_hidden",
"location": {"line": text[:match.start()].count("\n") + 1,
"context": content[:80]},
"raw_match": content[:100],
})
break
return findings
def detect_patterns(text, patterns, category):
"""通用模式匹配（3.4 修复：同行同模式去重）"""
findings = []
seen = set()
for p in patterns:
for match in re.finditer(p["pattern"], text, re.IGNORECASE):
line_num = text[:match.start()].count("\n") + 1
dedup_key = (category, p["name"], line_num)
if dedup_key in seen:
continue
seen.add(dedup_key)
lines = text.split("\n")
line_text = lines[line_num - 1] if line_num <= len(lines) else ""
findings.append({
"category": category,
"severity": p["severity"],
"pattern": p["name"],
"location": {"line": line_num, "context": line_text.strip()[:80]},
"raw_match": match.group()[:100],
})
return findings
def detect_typosquat(name, known_skills_path):
"""检测名称是否与已知 skill 过于相似"""
findings = []
if not name or not known_skills_path:
return findings
known_path = Path(known_skills_path)
if not known_path.exists():
return findings
known_names = []
for line in known_path.read_text().splitlines():
line = line.strip()
if line and not line.startswith("#"):
known_names.append(line.lower())
name_lower = name.lower()
if name_lower in known_names:
return findings
for known in known_names:
dist = _edit_distance(name_lower, known)
if 0 < dist <= 2 and len(name_lower) > 3:
findings.append({
"category": "typosquat",
"severity": "high",
"pattern": "edit_distance",
"location": {"line": 0, "context": f"名称 '{name}' 与已知 skill '{known}' 相似"},
"raw_match": f"编辑距离: {dist}",
})
break
homoglyphs = {"1": "l", "l": "1", "0": "o", "o": "0", "rn": "m", "m": "rn"}
for old, new in homoglyphs.items():
variant = name_lower.replace(old, new)
if variant != name_lower and variant in known_names:
findings.append({
"category": "typosquat",
"severity": "high",
"pattern": "homoglyph",
"location": {"line": 0, "context": f"名称 '{name}' 可能是 '{variant}' 的同形字变体"},
"raw_match": f"{old} -> {new}",
})
break
return findings
def _edit_distance(s1, s2):
"""Levenshtein 编辑距离"""
if len(s1) < len(s2):
return _edit_distance(s2, s1)
if len(s2) == 0:
return len(s1)
prev = range(len(s2) + 1)
for i, c1 in enumerate(s1):
curr = [i + 1]
for j, c2 in enumerate(s2):
curr.append(min(prev[j + 1] + 1, curr[j] + 1, prev[j] + (c1 != c2)))
prev = curr
return prev[len(s2)]
def decode_url_encoded(text):
"""检测并解码 URL 编码内容"""
findings = []
url_pattern = re.compile(r"(?:%[0-9A-Fa-f]{2}){3,}")
for match in url_pattern.finditer(text):
try:
from urllib.parse import unquote
decoded = unquote(match.group())
if any(re.search(p["pattern"], decoded, re.IGNORECASE) for p in INJECTION_PATTERNS + DANGEROUS_PATTERNS):
findings.append({
"category": "obfuscation",
"severity": "high",
"pattern": "url_encoded_injection",
"location": {"line": text[:match.start()].count("\n") + 1,
"context": match.group()[:60]},
"raw_match": f"URL decoded: {decoded[:100]}",
})
except Exception:
pass
return findings
def assess_permissions(text):
"""评估文件中暗示的权限范围"""
findings = []
has_network = bool(re.search(r"\b(fetch|curl|wget|axios|request)\s*\(", text, re.IGNORECASE)) or \
bool(re.search(r"\b(curl|wget)\s+\S", text, re.IGNORECASE))
has_shell = bool(re.search(r"\b(shell|bash|exec|spawn|child_process|subprocess)\b", text, re.IGNORECASE))
if has_network and has_shell:
findings.append({
"category": "permission_scope",
"severity": "high",
"pattern": "network_plus_shell",
"location": {"line": 0, "context": "同时需要 network + shell 权限（数据外传风险）"},
"raw_match": "network + shell combination detected",
})
elif has_shell:
findings.append({
"category": "permission_scope",
"severity": "medium",
"pattern": "shell_access",
"location": {"line": 0, "context": "需要 shell 权限"},
"raw_match": "shell access detected",
})
elif has_network:
findings.append({
"category": "permission_scope",
"severity": "low",
"pattern": "network_access",
"location": {"line": 0, "context": "需要 network 权限"},
"raw_match": "network access detected",
})
return findings
def scan_file(file_path, check_name=None, known_skills_path=None):
"""扫描单个文件，返回结果 dict"""
path = Path(file_path)
# 4.1 修复：文件不存在时优雅处理
if not path.exists():
return {
"scan_target": str(file_path),
"error": f"文件不存在: {file_path}",
"findings": [],
"summary": {"critical": 0, "high": 0, "medium": 0, "low": 0, "clean": True},
}
# 大小限制
file_size = path.stat().st_size
if file_size > MAX_SCAN_SIZE:
text = path.read_text(encoding="utf-8", errors="ignore")[:MAX_SCAN_SIZE]
else:
text = path.read_text(encoding="utf-8", errors="ignore")
# 2.2 修复：Unicode 同形字归一化（在原文上检测零宽字符后再归一化）
raw_text = text
text = normalize_unicode(text)
all_findings = []
# 1. Prompt injection
all_findings.extend(detect_patterns(text, INJECTION_PATTERNS, "prompt_injection"))
# 2. 危险命令
all_findings.extend(detect_patterns(text, DANGEROUS_PATTERNS, "dangerous_command"))
# 3. 数据外传
all_findings.extend(detect_patterns(text, DATA_EXFIL_PATTERNS, "data_exfiltration"))
# 4. 凭证访问
all_findings.extend(detect_patterns(text, CREDENTIAL_PATTERNS, "credential_access"))
# 5. CSS/样式隐藏
all_findings.extend(detect_patterns(text, OBFUSCATION_PATTERNS, "obfuscation"))
# 6. Base64 混淆
all_findings.extend(decode_base64_strings(text))
# 7. URL 编码混淆
all_findings.extend(decode_url_encoded(text))
# 8. 零宽字符（用原始文本检测，归一化前）
all_findings.extend(detect_zero_width(raw_text))
# 9. HTML 隐藏内容
all_findings.extend(detect_html_hidden(text))
# 10. 权限范围评估
all_findings.extend(assess_permissions(text))
# 11. Typosquat
if check_name:
all_findings.extend(detect_typosquat(check_name, known_skills_path))
# 汇总
severity_counts = {"critical": 0, "high": 0, "medium": 0, "low": 0}
for f in all_findings:
sev = f["severity"]
if sev in severity_counts:
severity_counts[sev] += 1
return {
"scan_target": str(file_path),
"findings": all_findings,
"summary": {
**severity_counts,
"clean": len(all_findings) == 0,
},
}
def main():
parser = argparse.ArgumentParser(description="安全扫描脚本")
parser.add_argument("file", nargs="?", help="要扫描的文件路径")
parser.add_argument("--check-name", help="要检查 typosquat 的名称")
parser.add_argument("--known-skills", help="known_skills.txt 路径")
parser.add_argument("--version", action="version", version=f"security_scan.py {__version__}")
args = parser.parse_args()
if not args.file:
parser.error("必须指定要扫描的文件路径")
result = scan_file(args.file, args.check_name, args.known_skills)
print(json.dumps(result, ensure_ascii=False, indent=2))
# 2.3 修复：exit code 区分 clean/dirty
if result["summary"]["critical"] > 0 or result["summary"]["high"] > 0:
sys.exit(2)
if __name__ == "__main__":
main()
Comprehensive crypto market scanner across Binance, OKX, Bybit, and Bitget. 12 scan types covering arbitrage (funding rate, basis, spot spread, futures sprea...
---
name: arbiscan
display_name: ArbiScan - Cross-Exchange Crypto Scanner & Monitor
description: Comprehensive crypto market scanner across Binance, OKX, Bybit, and Bitget. 12 scan types covering arbitrage (funding rate, basis, spot spread, futures spread), market monitoring (open interest, price movers, volume anomaly, stablecoin depeg, funding extreme), and trading signals (funding trend, long/short ratio, new listing detection). Read-only — no trading, no API keys needed.
version: 0.2.0
author: ZadAnthony
tags:
- crypto
- arbitrage
- defi
- trading
- scanner
- funding-rate
- basis
- spread
- monitoring
- open-interest
- volume
- signals
composable_with:
- binance/spot-trading
- binance/futures-trading
- bybit/trading
- bitget/trading
- coinank/funding-rate
- tradeos/executor
---
# ArbiScan — Cross-Exchange Crypto Scanner & Monitor
You are a comprehensive crypto market scanner. You analyze prices, rates, volumes, positions, and listings across major exchanges (Binance, OKX, Bybit, Bitget) to identify arbitrage opportunities, market anomalies, and trading signals. You only scan and report — you never execute trades.
## Symbol Format Reference
**CRITICAL: Each exchange uses different symbol formats. Always use the correct format for each exchange.**
| Exchange | Spot Format | Perpetual Format | Example Spot | Example Perp |
|----------|-------------|------------------|--------------|--------------|
| Binance | `{BASE}USDT` | `{BASE}USDT` | `BTCUSDT` | `BTCUSDT` |
| Bybit | `{BASE}USDT` | `{BASE}USDT` | `BTCUSDT` | `BTCUSDT` |
| OKX | `{BASE}-USDT` | `{BASE}-USDT-SWAP` | `BTC-USDT` | `BTC-USDT-SWAP` |
| Bitget | `{BASE}USDT` | `{BASE}USDT` | `BTCUSDT` | `BTCUSDT` |
## Error Handling
- If an exchange returns an error or times out, **skip it and continue** with data from other exchanges. Never abort a scan because one exchange is unavailable.
- If a symbol does not exist on an exchange (404 or empty response), skip that symbol on that exchange silently.
- Always present whatever data was successfully retrieved. Partial results are better than no results.
## User Intent Mapping
When the user asks a general question, map it to the appropriate scan(s):
| User says... | Run scan(s) |
|-------------|-------------|
| "套利机会" / "arbitrage opportunities" | funding + basis + spread |
| "市场怎么样" / "market overview" | price_movers + volume_anomaly + funding_extreme |
| "XX币怎么样" / "how is BTC doing" | funding + basis + price_movers + long_short (for that symbol only) |
| "费率" / "funding rate" | funding (+ funding_history for deeper analysis) |
| "稳定币" / "stablecoins" | depeg |
| "新币" / "new listings" | new_listing |
| "风险" / "risk" / "危险信号" | funding_extreme + long_short + open_interest |
| "全扫" / "scan everything" | all 12 scans |
When the user specifies a particular symbol (e.g., "scan TRUMP"), only scan that symbol — do not scan all 100.
---
## Capabilities: 12 Scan Types
## Category A: Arbitrage Scans
### 1. Funding Rate Arbitrage
Compare perpetual contract funding rates across exchanges. Long on the cheap side, short on the expensive side.
**Workflow:**
1. For each symbol, fetch current funding rates from all 4 exchanges
2. Find lowest and highest rates across exchanges
3. Calculate rate difference and annualized yield: `APY = rate_diff × (365 × 24 / interval_hours) × 100`
4. Filter by minimum APY threshold (default: 0%)
5. Assign risk level: LOW (major coin + APY<10%), MEDIUM (APY>10%, or any non-major coin regardless of APY), HIGH (APY>50%, or non-major + APY>20%). Major coins: BTC, ETH, BNB, SOL, XRP
6. Output sorted by APY descending
**API endpoints and response parsing:**
Binance:
```
GET https://fapi.binance.com/fapi/v1/premiumIndex?symbol=BTCUSDT
Response: { "lastFundingRate": "0.00010000", ... }
→ Read: float(response["lastFundingRate"])
```
Bybit:
```
GET https://api.bybit.com/v5/market/tickers?category=linear&symbol=BTCUSDT
Response: { "result": { "list": [{ "fundingRate": "0.0001", ... }] } }
→ Read: float(response["result"]["list"][0]["fundingRate"])
```
OKX:
```
GET https://www.okx.com/api/v5/public/funding-rate?instId=BTC-USDT-SWAP
Response: { "data": [{ "fundingRate": "0.0001", ... }] }
→ Read: float(response["data"][0]["fundingRate"])
```
Bitget:
```
GET https://api.bitget.com/api/v2/mix/market/current-fund-rate?symbol=BTCUSDT&productType=USDT-FUTURES
Response: { "data": [{ "fundingRate": "0.0001", ... }] }
→ Read: float(response["data"][0]["fundingRate"])
```
### 2. Basis Arbitrage (Spot vs Futures)
Compare spot and perpetual futures prices on the same exchange.
**Workflow:**
1. Fetch spot and futures prices from each exchange for each symbol
2. Calculate basis: `(futures - spot) / spot × 100`
3. Flag Contango (futures > spot) or Backwardation (futures < spot)
4. Filter by absolute basis threshold (default: 0.05%)
**API endpoints — Spot price:**
Binance:
```
GET https://api.binance.com/api/v3/ticker/price?symbol=BTCUSDT
→ Read: float(response["price"])
```
Bybit:
```
GET https://api.bybit.com/v5/market/tickers?category=spot&symbol=BTCUSDT
→ Read: midpoint of float(response["result"]["list"][0]["bid1Price"]) and ["ask1Price"]
```
OKX:
```
GET https://www.okx.com/api/v5/market/ticker?instId=BTC-USDT
→ Read: midpoint of float(response["data"][0]["bidPx"]) and ["askPx"]
```
Bitget:
```
GET https://api.bitget.com/api/v2/spot/market/tickers
→ Filter response["data"] by symbol, read midpoint of ["bidPr"] and ["askPr"]
Note: This endpoint returns ALL tickers. Filter by matching symbol field.
```
**API endpoints — Futures price:**
Binance:
```
GET https://fapi.binance.com/fapi/v1/ticker/price?symbol=BTCUSDT
→ Read: float(response["price"])
```
Bybit:
```
GET https://api.bybit.com/v5/market/tickers?category=linear&symbol=BTCUSDT
→ Read: float(response["result"]["list"][0]["lastPrice"])
```
OKX:
```
GET https://www.okx.com/api/v5/market/ticker?instId=BTC-USDT-SWAP
→ Read: float(response["data"][0]["last"])
```
Bitget:
```
GET https://api.bitget.com/api/v2/mix/market/tickers?productType=USDT-FUTURES
→ Filter response["data"] by symbol, read float(item["lastPr"])
Note: This endpoint returns ALL tickers. Filter by matching symbol field.
```
### 3. Cross-Exchange Spot Spread
Compare bid/ask prices across exchanges for the same symbol.
**Workflow:**
1. Fetch best bid/ask from all exchanges (use spot ticker endpoints from scan #2)
2. Compare all exchange pairs: if bid_B > ask_A, spread exists
3. Calculate spread percentage: `(bid_B - ask_A) / ask_A × 100`
4. Filter by minimum spread threshold (default: 0.02%)
**API endpoints — Spot bid/ask:**
Binance:
```
GET https://api.binance.com/api/v3/ticker/bookTicker?symbol=BTCUSDT
→ Read: bid = float(response["bidPrice"]), ask = float(response["askPrice"])
```
Bybit:
```
GET https://api.bybit.com/v5/market/tickers?category=spot&symbol=BTCUSDT
→ Read: bid = float(response["result"]["list"][0]["bid1Price"]), ask = ["ask1Price"]
```
OKX:
```
GET https://www.okx.com/api/v5/market/ticker?instId=BTC-USDT
→ Read: bid = float(response["data"][0]["bidPx"]), ask = ["askPx"]
```
Bitget:
```
GET https://api.bitget.com/api/v2/spot/market/tickers
→ Filter by symbol, read bid = float(item["bidPr"]), ask = float(item["askPr"])
```
### 4. Cross-Exchange Futures Spread
Compare perpetual contract prices across exchanges pairwise.
**Workflow:**
1. Fetch futures prices from all exchanges for each symbol (use futures price endpoints from scan #2)
2. Compare all exchange pairs pairwise (up to 6 pairs for 4 exchanges)
3. Calculate spread percentage: `(high_price - low_price) / low_price × 100`
4. Filter by minimum spread threshold (default: 0.03%)
5. A single symbol may appear multiple times (once per exchange pair that exceeds threshold)
---
## Category B: Market Monitoring
### 5. Stablecoin Depeg Monitor
Monitor stablecoin prices for deviation from $1.00 (quoted in USDT).
**Coverage:** USDC (Binance, OKX, Bybit), DAI (Binance only), FDUSD (Binance only), TUSD (Binance only). Bitget is not covered for this scan.
**Workflow:**
1. Fetch stablecoin prices from available exchanges
2. Calculate deviation from $1.00
3. Flag: STABLE (<0.1%), WATCH (0.1-0.5%), DEPEGGED (>0.5%)
**API endpoints:**
```
Binance: GET https://api.binance.com/api/v3/ticker/price?symbol=USDCUSDT → float(response["price"])
OKX: GET https://www.okx.com/api/v5/market/ticker?instId=USDC-USDT → float(response["data"][0]["last"])
Bybit: GET https://api.bybit.com/v5/market/tickers?category=spot&symbol=USDCUSDT → float(response["result"]["list"][0]["lastPrice"])
```
Replace USDC with DAI, FDUSD, TUSD as needed (Binance only for those).
### 6. Open Interest Monitor
Track open interest distribution across exchanges. Shows concentration levels for each symbol.
**Workflow:**
1. Fetch open interest from all exchanges for each symbol (default: Top 30 symbols)
2. Compare OI across exchanges, calculate each exchange's share
3. Label: CONCENTRATED (>60% on one exchange), MODERATE (45-60%), BALANCED (<45%)
4. Output all symbols with data from ≥2 exchanges, sorted by concentration
**API endpoints and response parsing:**
Binance:
```
GET https://fapi.binance.com/fapi/v1/openInterest?symbol=BTCUSDT
→ Read: float(response["openInterest"]) (unit: contracts in base asset)
```
Bybit:
```
GET https://api.bybit.com/v5/market/open-interest?category=linear&symbol=BTCUSDT&intervalTime=5min
→ Read: float(response["result"]["list"][0]["openInterest"])
```
OKX:
```
GET https://www.okx.com/api/v5/public/open-interest?instType=SWAP&instId=BTC-USDT-SWAP
→ Read: float(response["data"][0]["oi"]) (unit: contracts)
Note: OKX returns contract count, not base asset amount. Units differ from other exchanges.
```
Bitget:
```
GET https://api.bitget.com/api/v2/mix/market/open-interest?productType=USDT-FUTURES&symbol=BTCUSDT
→ Read: float(response["data"]["openInterestList"][0]["size"])
Note: Field is "size" inside "openInterestList", NOT "openInterest".
```
### 7. Funding Rate Extreme Alert
Flag symbols with extreme funding rates (> ±0.1%) that signal overcrowded positioning and potential reversal.
**Workflow:**
1. Fetch current funding rates from all exchanges (same endpoints as scan #1)
2. Flag any rate exceeding ±0.1% (normal is ~0.01%)
3. Calculate multiples above normal: `multiple = abs(rate) / 0.0001`
4. Label direction: rate > 0 → "LONG CROWDED", rate < 0 → "SHORT CROWDED"
5. Higher extremes = higher reversal probability
### 8. Price Movers (24h Gainers/Losers)
Identify the biggest price movers in the last 24 hours across all exchanges.
**Workflow:**
1. Fetch 24h ticker data from all exchanges
2. Extract price change percentage
3. Rank by absolute change (combined list of gainers and losers, not separate)
4. Deduplicate by symbol (keep the exchange with the largest move), show top N (default: 20)
**API endpoints and response parsing:**
Binance:
```
GET https://fapi.binance.com/fapi/v1/ticker/24hr?symbol=BTCUSDT
→ Read: change = float(response["priceChangePercent"]), volume = float(response["quoteVolume"]), price = float(response["lastPrice"])
```
Bybit:
```
GET https://api.bybit.com/v5/market/tickers?category=linear&symbol=BTCUSDT
→ Read: price = float(["lastPrice"]), prev = float(["prevPrice24h"])
→ Compute: change = (price - prev) / prev × 100
→ Volume: float(["turnover24h"])
```
OKX:
```
GET https://www.okx.com/api/v5/market/ticker?instId=BTC-USDT-SWAP
→ Read: last = float(["last"]), open = float(["open24h"])
→ Compute: change = (last - open) / open × 100
→ Volume: float(["volCcy24h"])
```
Bitget:
```
GET https://api.bitget.com/api/v2/mix/market/tickers?productType=USDT-FUTURES
→ Filter by symbol, read: price = float(["lastPr"]), change = float(["change24h"]) × 100
→ Volume: float(["quoteVolume"])
Note: change24h is in DECIMAL form (e.g., 0.05 = 5%). Multiply by 100.
Note: Returns ALL tickers. Filter by matching symbol field.
```
### 9. Volume Anomaly Detection
Detect symbols with unusual volume distribution across exchanges.
**Workflow:**
1. Fetch 24h volume from all exchanges for each symbol (same endpoints as scan #8)
2. Calculate each exchange's share of total volume
3. Assign signal: VOLUME SPIKE (>70% share on one exchange), ACCUMULATION (>50% share + price change <2%), MOMENTUM (price change >10% regardless of volume distribution), NORMAL (filtered out)
4. Only non-NORMAL results are shown in output
---
## Category C: Trading Signals
### 10. Funding Rate Trend
Analyze historical funding rates to find persistent patterns. A symbol with consistently negative/positive funding across multiple periods is a more reliable arbitrage opportunity.
**Workflow:**
1. Fetch last 20 funding rate records for each symbol (default: Top 30 symbols)
2. Count how many consecutive periods the rate stayed positive/negative
3. Calculate average rate over the streak period
4. Flag symbols with ≥5 consecutive same-direction rates as "trending"
5. Annualize the average rate for yield estimate
6. Label stability: EMERGING (5-9 periods), STABLE (10-14), VERY STABLE (15+)
**API endpoints (Binance, Bybit, OKX only — Bitget not included in this scan):**
Binance:
```
GET https://fapi.binance.com/fapi/v1/fundingRate?symbol=BTCUSDT&limit=20
→ Response: list of { "fundingRate": "0.0001", ... } (oldest first)
→ Read: [float(item["fundingRate"]) for item in reversed(response)]
```
Bybit:
```
GET https://api.bybit.com/v5/market/funding/history?category=linear&symbol=BTCUSDT&limit=20
→ Read: [float(item["fundingRate"]) for item in response["result"]["list"]] (newest first)
```
OKX:
```
GET https://www.okx.com/api/v5/public/funding-rate-history?instId=BTC-USDT-SWAP&limit=20
→ Read: [float(item["fundingRate"]) for item in response["data"]]
```
### 11. Long/Short Ratio
Track the ratio of long vs short positions. Extreme ratios (e.g., 80% long) often precede reversals.
**Workflow:**
1. Fetch global long/short account ratio (default: Top 30 symbols)
2. Include all ratios where either side exceeds 60%
3. Label signals: EXTREME LONG/SHORT (>75%), LONG/SHORT HEAVY (>65%), MODERATE (>60%)
4. Output sorted by ratio extremity
**API endpoints (Binance and Bybit only — OKX and Bitget do not expose this data):**
Binance:
```
GET https://fapi.binance.com/futures/data/globalLongShortAccountRatio?symbol=BTCUSDT&period=5m&limit=1
→ Response: [{ "longShortRatio": "1.5", ... }]
→ Compute: ratio = float(response[0]["longShortRatio"])
→ long_pct = ratio / (1 + ratio) × 100, short_pct = 100 - long_pct
```
Bybit:
```
GET https://api.bybit.com/v5/market/account-ratio?category=linear&symbol=BTCUSDT&period=1h&limit=1
→ Read: buyRatio = float(response["result"]["list"][0]["buyRatio"]), sellRatio = ["sellRatio"]
→ long_pct = buyRatio / (buyRatio + sellRatio) × 100
```
### 12. New Listing Detection
Compare trading pair lists across exchanges to find tokens available on some but not all exchanges.
**Workflow:**
1. Fetch full list of USDT trading pairs from all 4 exchanges
2. Compare: find symbols present on 1-2 exchanges but missing on others
3. Label exclusivity: EXCLUSIVE (1 exchange only), LIMITED (2 exchanges) — note: no recency/date information available
4. Show which exchanges have it and which don't
**API endpoints and response parsing:**
Binance:
```
GET https://api.binance.com/api/v3/exchangeInfo
→ Filter: [s["symbol"] for s in response["symbols"] if s["quoteAsset"] == "USDT" and s["status"] == "TRADING"]
```
Bybit:
```
GET https://api.bybit.com/v5/market/instruments-info?category=spot
→ Filter: [s["symbol"] for s in response["result"]["list"] if s["symbol"].endswith("USDT") and s["status"] == "Trading"]
```
OKX:
```
GET https://www.okx.com/api/v5/public/instruments?instType=SPOT
→ Filter: [s["instId"].replace("-","") for s in response["data"] if s["instId"].endswith("-USDT") and s["state"] == "live"]
```
Bitget:
```
GET https://api.bitget.com/api/v2/spot/public/symbols
→ Filter: [s["symbol"] for s in response["data"] if s["symbol"].endswith("USDT") and s["status"] == "online"]
```
---
## Output Format
Always present results in a clear table appropriate to the scan type. Example for funding rate:
```
| Symbol | Long (低费率) | Short (高费率) | Rate Diff | Est. APY | Risk | Window |
|---------|---------------|---------------|-----------|----------|--------|--------|
| ETHUSDT | Bybit 0.001% | Binance 0.05% | 0.049% | 53.7% | MEDIUM | ~8h |
```
## Scan Categories Quick Reference
| Category | Scans | Purpose |
|----------|-------|---------|
| **Arbitrage** | funding, basis, spread, futures_spread | Find price/rate discrepancies to exploit |
| **Monitoring** | depeg, open_interest, funding_extreme, price_movers, volume_anomaly | Track market conditions and anomalies |
| **Signals** | funding_history, long_short, new_listing | Identify trading signals and opportunities |
## Important Notes
- **Read-only**: This skill only scans and reports. It never places orders or moves funds.
- **No API keys needed**: All data comes from public endpoints.
- **Not financial advice**: Opportunities shown are theoretical. Actual execution requires considering gas fees, withdrawal times, slippage, and exchange risks.
- **Rate limits**: Respect exchange rate limits. Add ~200ms delay between requests. Do not send burst requests.
- **Coverage**: Default Top 30 trading pairs by market cap. Users can request broader scans (e.g., "scan Top 100" or "scan all coins") — the agent can scan any symbol that exists on the exchanges. 4 major exchanges (some scans have limited exchange coverage — see individual scan descriptions).
- **Bitget bulk endpoints**: Bitget's spot tickers and futures tickers endpoints return ALL symbols at once. Filter the response by symbol name rather than passing a symbol parameter.
## Composable Usage
ArbiScan works best when composed with exchange trading skills:
1. **ArbiScan** identifies opportunities and signals (this skill)
2. **Exchange Skills** (binance/bybit/bitget) can execute trades if the user decides to act
3. **TradeOS** can manage the full workflow
The user always makes the final decision on whether to act on any opportunity.
## Standalone Mode
ArbiScan ships with Python scripts that can run independently:
```bash
cd scripts/
pip install -r requirements.txt
# Run all scans
python scanner.py --all
# By category
python scanner.py --type arbitrage
python scanner.py --type monitor
python scanner.py --type signals
# Individual scans
python scanner.py --type funding --min-apy 10
python scanner.py --type price_movers
python scanner.py --type long_short
python scanner.py --type new_listing
# Output formats
python scanner.py --type funding --format markdown
python scanner.py --type price_movers --format json
```
FILE:README.md
# ArbiScan — Cross-Exchange Crypto Scanner & Monitor
[](https://opensource.org/licenses/MIT)
**12 scan types across Binance, OKX, Bybit, and Bitget. Arbitrage, monitoring, and signals — all from public APIs, no keys needed.**
ArbiScan is an [OpenClaw Skill](https://clawhub.ai) and standalone CLI that scans major crypto exchanges for arbitrage opportunities, market anomalies, and trading signals. It covers 100 trading pairs across 4 exchanges. You decide whether to act — ArbiScan only watches.
## Scan Types
### Arbitrage (find price/rate discrepancies)
| Type | What it does | Data Source |
|------|-------------|-------------|
| **Funding Rate Arb** | Compares perpetual funding rates across exchanges | Funding rate endpoints |
| **Basis Arb** | Spots premium/discount between spot and futures | Spot + futures tickers |
| **Spot Spread** | Finds bid/ask gaps across exchanges | Order book top-of-book |
| **Futures Spread** | Finds perpetual contract price gaps across exchanges | Futures tickers |
### Monitoring (track market conditions)
| Type | What it does | Data Source |
|------|-------------|-------------|
| **Stablecoin Depeg** | Monitors USDC/DAI/FDUSD/TUSD deviation from $1 | Stablecoin tickers |
| **Open Interest** | Tracks OI distribution, flags concentration on one exchange | OI endpoints |
| **Funding Extreme** | Alerts when funding rate exceeds ±0.1% (10x normal) | Funding rate endpoints |
| **Price Movers** | 24h top gainers and losers | 24hr tickers |
| **Volume Anomaly** | Detects unusual volume concentration or spikes | 24hr volume data |
### Signals (identify trading signals)
| Type | What it does | Data Source |
|------|-------------|-------------|
| **Funding Trend** | Finds symbols with ≥5 consecutive same-direction funding rates | Historical funding rates |
| **Long/Short Ratio** | Flags extreme positioning (>65% one-sided) | Binance + Bybit ratio endpoints |
| **New Listing** | Tokens on some exchanges but not others — premium potential | Exchange pair lists |
## Quick Start
### As an OpenClaw Skill
Install from ClawHub and let your AI agent scan:
```
"Scan for funding rate arbitrage opportunities with APY > 10%"
"Which coins have extreme funding rates right now?"
"Show me the biggest price movers in the last 24 hours"
"Are any stablecoins depegging?"
"Find tokens listed on Binance but not on OKX"
```
### Standalone (Python)
```bash
cd scripts/
pip install -r requirements.txt
# Run everything
python scanner.py --all
# Run by category
python scanner.py --type arbitrage # all 4 arbitrage scans
python scanner.py --type monitor # all 5 monitoring scans
python scanner.py --type signals # all 3 signal scans
# Individual scans
python scanner.py --type funding --min-apy 10
python scanner.py --type price_movers
python scanner.py --type long_short
python scanner.py --type new_listing
# Output formats
python scanner.py --type funding --format markdown
python scanner.py --type price_movers --format json
```
## Sample Output
```
Funding Rate Arbitrage
================================================================================
Symbol Long (低费率) Short (高费率) Rate Diff Est. APY% Risk Window
--------- ------------------- -------------------- ---------- ---------- ------ --------
FILUSDT Binance -0.0799% Bitget -0.0104% 0.0695% 76.1% HIGH ~8h
SEIUSDT Binance -0.0317% OKX -0.0026% 0.0291% 31.9% MEDIUM ~8h
BTCUSDT Binance -0.0027% OKX 0.0064% 0.0091% 10.0% LOW ~8h
```
```
24h Price Movers (Gainers & Losers)
================================================================================
Symbol Exchange Price 24h Change% 24h Volume (USDT) Direction
--------- ---------- ----------- ------------- ------------------- ---------
ARBUSDT Binance $0.1017 -2.59% $41,804,011 DUMP
SOLUSDT OKX $87.9300 -1.15% $11,760,759 DUMP
SEIUSDT Bybit $0.0664 +0.54% $15,798,499 PUMP
```
## Composable with Exchange Skills
ArbiScan is designed to work alongside exchange trading skills:
1. **ArbiScan** scans and identifies opportunities (this skill)
2. **Binance/Bybit/Bitget Skills** can execute trades if you decide to act
3. **TradeOS** can manage the full workflow
```
"Use ArbiScan to find opportunities, then use Binance skill to execute the best one"
```
## How It Works
- Fetches data from **public API endpoints only** — no API keys, no authentication
- Built-in rate limiting (200ms between requests) to respect exchange limits
- Covers **Top 30 trading pairs** by default (expandable on request) by market cap
- **12 scan types** across 3 categories (arbitrage, monitoring, signals)
- Risk scoring based on APY magnitude and coin category
## Covered Exchanges
| Exchange | Spot | Futures | Funding Rate | Open Interest | Long/Short Ratio |
|----------|------|---------|--------------|---------------|-----------------|
| Binance | ✅ | ✅ | ✅ | ✅ | ✅ |
| Bybit | ✅ | ✅ | ✅ | ✅ | ✅ |
| OKX | ✅ | ✅ | ✅ | ✅ | — |
| Bitget | ✅ | ✅ | ✅ | ✅ | — |
## Disclaimer
ArbiScan is for **informational purposes only**. It does not constitute financial advice. Opportunities shown are theoretical — actual execution requires considering:
- Gas/withdrawal fees
- Transfer times between exchanges
- Slippage and liquidity
- Exchange counterparty risk
- Regulatory compliance
Always do your own research before trading.
## License
MIT
FILE:README_CN.md
# ArbiScan — 跨交易所加密货币扫描 & 监控
[](https://opensource.org/licenses/MIT)
**12 种扫描，覆盖 Binance、OKX、Bybit、Bitget。套利、监控、信号 — 全部公开 API，无需 key。**
ArbiScan 是一个 [OpenClaw Skill](https://clawhub.ai) 和独立 CLI 工具，扫描主流交易所的套利机会、市场异常和交易信号。覆盖 100 个交易对，4 个交易所。是否行动由你决定 — ArbiScan 只负责发现。
## 扫描类型
### 套利类（发现价格/费率差异）
| 类型 | 功能 | 数据源 |
|------|------|--------|
| **资金费率套利** | 比较各交易所永续合约资金费率差异 | 资金费率端点 |
| **期现基差套利** | 发现同交易所现货与合约价差 | 现货 + 合约价格 |
| **跨所现货价差** | 寻找不同交易所之间的 bid/ask 价差 | 订单簿最优报价 |
| **跨所合约价差** | 寻找不同交易所之间的合约价差 | 合约行情 |
### 监控类（跟踪市场状态）
| 类型 | 功能 | 数据源 |
|------|------|--------|
| **稳定币脱锚** | 监测 USDC/DAI/FDUSD/TUSD 偏离 $1 | 稳定币价格 |
| **持仓量监控** | 跟踪 OI 分布，标记某交易所集中度过高 | 持仓量端点 |
| **费率异常警报** | 费率超过 ±0.1%（正常值 10 倍）时告警 | 资金费率端点 |
| **24h 涨跌幅排行** | 最近 24 小时涨跌最大的币种 | 24h 行情 |
| **成交量异常** | 检测某交易所成交量异常集中或飙升 | 24h 成交量 |
### 信号类（识别交易信号）
| 类型 | 功能 | 数据源 |
|------|------|--------|
| **费率趋势** | 连续 ≥5 期同方向费率 = 稳定套利机会 | 历史费率 |
| **多空比** | 标记极端仓位（一方 >65%）| Binance + Bybit 多空比 |
| **新币上线检测** | A 所有 B 所没有的币 = 溢价窗口 | 各所交易对列表 |
## 快速开始
### 作为 OpenClaw Skill
从 ClawHub 安装，让 AI Agent 帮你扫描：
```
"扫描资金费率套利机会，APY 大于 10%"
"哪些币的费率现在很极端？"
"看看过去 24 小时涨跌最大的币"
"有没有稳定币在脱锚？"
"找出只在 Binance 上线但 OKX 没有的币"
```
### 独立运行（Python）
```bash
cd scripts/
pip install -r requirements.txt
# 运行所有扫描
python scanner.py --all
# 按分类运行
python scanner.py --type arbitrage # 4 种套利扫描
python scanner.py --type monitor # 5 种监控扫描
python scanner.py --type signals # 3 种信号扫描
# 单独运行
python scanner.py --type funding --min-apy 10
python scanner.py --type price_movers
python scanner.py --type long_short
python scanner.py --type new_listing
# 输出格式
python scanner.py --type funding --format markdown
python scanner.py --type price_movers --format json
```
## 示例输出
```
Funding Rate Arbitrage
================================================================================
Symbol Long (低费率) Short (高费率) Rate Diff Est. APY% Risk Window
--------- ------------------- -------------------- ---------- ---------- ------ --------
FILUSDT Binance -0.0799% Bitget -0.0104% 0.0695% 76.1% HIGH ~8h
SEIUSDT Binance -0.0317% OKX -0.0026% 0.0291% 31.9% MEDIUM ~8h
BTCUSDT Binance -0.0027% OKX 0.0064% 0.0091% 10.0% LOW ~8h
```
```
24h Price Movers (Gainers & Losers)
================================================================================
Symbol Exchange Price 24h Change% 24h Volume (USDT) Direction
--------- ---------- ----------- ------------- ------------------- ---------
ARBUSDT Binance $0.1017 -2.59% $41,804,011 DUMP
SOLUSDT OKX $87.9300 -1.15% $11,760,759 DUMP
SEIUSDT Bybit $0.0664 +0.54% $15,798,499 PUMP
```
## 可组合使用
ArbiScan 设计为与交易所交易 Skill 配合使用：
1. **ArbiScan** 扫描发现机会和信号（本 Skill）
2. **Binance/Bybit/Bitget Skill** 执行交易（如果你决定行动）
3. **TradeOS** 可管理完整工作流
```
"用 ArbiScan 找机会，然后用 Binance skill 执行最优的那个"
```
## 设计理念："看和用分开"
ArbiScan **只扫描、不交易**：
- 零风险：不接触你的资金，不需要 API key
- 纯信息：发现机会后，执行交给你或交易所 Skill
- 可组合：和 Binance/Bybit/Bitget 的交易 Skill 配合使用
## 工作原理
- 仅使用**公开 API 端点** — 无需 API key，无需认证
- 内置限频（请求间隔 200ms），遵守交易所速率限制
- 覆盖市值 **Top 30 交易对**（可按需扩展）
- **12 种扫描**，分 3 大类（套利、监控、信号）
- 基于年化收益和币种类别的风险评分
## 支持的交易所
| 交易所 | 现货 | 合约 | 资金费率 | 持仓量 | 多空比 |
|--------|------|------|----------|--------|--------|
| Binance | ✅ | ✅ | ✅ | ✅ | ✅ |
| Bybit | ✅ | ✅ | ✅ | ✅ | ✅ |
| OKX | ✅ | ✅ | ✅ | ✅ | — |
| Bitget | ✅ | ✅ | ✅ | ✅ | — |
## 免责声明
ArbiScan 仅供信息参考，不构成投资建议。显示的机会是理论值，实际执行需考虑：
- Gas/提币手续费
- 交易所之间的转账时间
- 滑点和流动性
- 交易所对手风险
- 合规要求
请自行研究后再做决策。
## 开源协议
MIT
FILE:examples/sample_output.md
# ArbiScan v0.2.0 Sample Output
> Scanned at 2026-03-14 05:30 UTC | 12 scan types | 4 exchanges
## Arbitrage Scans
### Funding Rate Arbitrage
| Symbol | Long (低费率) | Short (高费率) | Rate Diff | Est. APY% | Risk | Window |
|-----------|---------------------|---------------------|-------------|-------------|--------|--------|
| FILUSDT | Binance -0.0799% | Bitget -0.0104% | 0.0695% | 76.1% | HIGH | ~8h |
| SEIUSDT | Binance -0.0317% | OKX -0.0026% | 0.0291% | 31.9% | MEDIUM | ~8h |
| BTCUSDT | Binance -0.0027% | OKX 0.0064% | 0.0091% | 10.0% | LOW | ~8h |
| SOLUSDT | Binance -0.0103% | Bybit -0.0004% | 0.0100% | 10.9% | LOW | ~8h |
> 28 opportunities found
### Basis Arbitrage (Spot vs Futures)
| Symbol | Exchange | Spot Price | Futures Price | Basis% | Direction | Risk |
|-----------|----------|------------|---------------|----------|---------------|--------|
| WIFUSDT | Binance | $0.16 | $0.16 | -0.5488% | Backwardation | MEDIUM |
| LINKUSDT | Binance | $8.93 | $8.91 | -0.1904% | Backwardation | MEDIUM |
| ETHUSDT | Binance | $2021.11 | $2020.06 | -0.0520% | Backwardation | LOW |
> 25 opportunities found
### Cross-Exchange Futures Spread
| Symbol | Buy From | Sell To | Low Price | High Price | Spread% | Risk |
|-----------|-------------------|--------------------|------------|------------|---------|--------|
| LTCUSDT | Bitget $53.89 | Binance $53.96 | $53.89 | $53.96 | 0.1298% | MEDIUM |
| BNBUSDT | OKX $641.00 | Binance $641.50 | $641.00 | $641.50 | 0.0780% | LOW |
> 24 opportunities found
## Monitoring Scans
### Open Interest Monitor
| Symbol | Total OI (USDT) | Top Exchange | Share% | Status |
|-----------|-----------------|------------------|--------|--------------|
| BTCUSDT | $2,979,093 | OKX (95.6%) | 95.6% | CONCENTRATED |
| DOGEUSDT | $2,935,208,407 | Binance (64.4%) | 64.4% | CONCENTRATED |
| SOLUSDT | $18,601,879 | Binance (51.4%) | 51.4% | MODERATE |
### 24h Price Movers
| Symbol | Exchange | Price | 24h Change% | 24h Volume (USDT) | Direction |
|-----------|----------|-------------|-------------|-------------------|-----------|
| ARBUSDT | Binance | $0.1017 | -2.59% | $41,804,011 | DUMP |
| FILUSDT | Binance | $0.8700 | -2.47% | $103,375,262 | DUMP |
| SEIUSDT | Bybit | $0.0664 | +0.54% | $15,798,499 | PUMP |
### Volume Anomaly Detection
| Symbol | Total Vol (USDT) | Top Exchange | Vol Share% | 24h Change% | Signal |
|-----------|------------------|--------------|------------|-------------|--------------|
| DOGEUSDT | $2,100,000,000 | OKX | 85.5% | -1.10% | VOLUME SPIKE |
| BTCUSDT | $25,000,000,000 | Binance | 58.2% | -0.78% | ACCUMULATION |
### Stablecoin Depeg Monitor
| Stablecoin | Exchange | Price (USDT) | Depeg% | Status |
|------------|----------|--------------|---------|--------|
| TUSD | Binance | $0.999700 | -0.0300%| STABLE |
| USDC | Binance | $1.000000 | +0.0000%| STABLE |
## Signal Scans
### Funding Rate History Trend
| Symbol | Exchange | Streak | Direction | Avg Rate | Est. APY% | Stability |
|-----------|----------|------------|-----------|----------|-----------|------------|
| BTCUSDT | Binance | 20 periods | NEGATIVE | -0.0050% | 5.5% | VERY STABLE|
| ETHUSDT | OKX | 12 periods | POSITIVE | 0.0080% | 8.8% | STABLE |
| FILUSDT | Binance | 7 periods | NEGATIVE | -0.0900% | 98.6% | EMERGING |
> 22 trending symbols found
### Long/Short Ratio
| Symbol | Exchange | Long% | Short% | Ratio | Signal |
|-----------|----------|--------|--------|-------|-------------------------------|
| WIFUSDT | Binance | 72.3% | 27.7% | 2.61 | LONG HEAVY |
| DOGEUSDT | Bybit | 31.5% | 68.5% | 0.46 | SHORT HEAVY |
| BTCUSDT | Binance | 54.5% | 45.5% | 1.20 | MODERATE |
> 41 extreme ratios found
### New Listing Detection
| Symbol | Available On | Missing From | Exclusivity | Signal |
|--------------|------------------|---------------------------|-------------|----------------------------|
| NEWTOKENUSDT | Bitget | Binance, Bybit, OKX | 1/4 | EXCLUSIVE - high premium |
| ALPHATOKEN | Binance, Bybit | OKX, Bitget | 2/4 | LIMITED - moderate premium |
> 522 exclusive/limited listings found
## How to Read This
- **Risk**: LOW (major coin + APY<10%), MEDIUM (APY 10-50% or non-major), HIGH (APY>50% or non-major + APY>20%)
- **Status** (OI): BALANCED (<45%), MODERATE (45-60%), CONCENTRATED (>60%)
- **Signal** (Volume): NORMAL, ACCUMULATION (high vol + low change), VOLUME SPIKE (>70% share), MOMENTUM (high change)
- **Stability** (Funding Trend): EMERGING (5-9 periods), STABLE (10-14), VERY STABLE (15+)
- **Exclusivity** (New Listing): 1/4 = only on 1 exchange, 2/4 = on 2 exchanges
## Disclaimer
These are **theoretical opportunities**. Actual returns depend on execution costs, slippage, transfer times, and exchange risks. This is not financial advice.
FILE:scripts/basis_arb.py
"""期现基差套利扫描 — 同所 spot vs futures 价差"""
from config import TOP_SYMBOLS, EXCHANGES
from fetcher import fetch_all_spot_prices, fetch_all_futures_prices
from formatter import format_output, risk_level
def scan_basis_arbitrage(symbols: list = None, min_basis_pct: float = 0.05) -> tuple:
"""
扫描期现基差套利机会
返回 (rows, headers)
"""
if symbols is None:
symbols = TOP_SYMBOLS
headers = ["Symbol", "Exchange", "Spot Price", "Futures Price", "Basis%", "Direction", "Risk"]
rows = []
print(f"[Basis Arb] 扫描 {len(symbols)} 个交易对...")
for symbol in symbols:
spot_prices = fetch_all_spot_prices(symbol)
futures_prices = fetch_all_futures_prices(symbol)
# 找同一交易所的期现价差
common_exchanges = set(spot_prices.keys()) & set(futures_prices.keys())
for exchange in common_exchanges:
spot = spot_prices[exchange]
futures = futures_prices[exchange]
if spot == 0:
continue
basis_pct = (futures - spot) / spot * 100
if abs(basis_pct) < min_basis_pct:
continue
direction = "Contango" if basis_pct > 0 else "Backwardation"
risk = risk_level(abs(basis_pct) * 365, symbol) # 粗略年化
rows.append([
f"{symbol}USDT",
EXCHANGES[exchange]["name"],
f".2f",
f".2f",
f"{basis_pct:.4f}%",
direction,
risk,
])
rows.sort(key=lambda r: abs(float(r[4].rstrip('%'))), reverse=True)
return rows, headers
def main():
import argparse
parser = argparse.ArgumentParser(description="Basis Arbitrage Scanner")
parser.add_argument("--format", choices=["table", "markdown", "json"], default="table")
args = parser.parse_args()
rows, headers = scan_basis_arbitrage()
print(f"\n{'='*80}")
print(" Basis Arbitrage (Spot vs Futures)")
print(f"{'='*80}")
print(format_output(rows, headers, args.format))
print(f"\nFound {len(rows)} opportunities")
if __name__ == "__main__":
main()
FILE:scripts/config.py
"""交易所端点配置 + symbol 映射"""
# 默认 Top 30 交易对（按市值排序，用户可通过 Agent 指定扫描更多）
TOP_SYMBOLS = [
"BTC", "ETH", "BNB", "SOL", "XRP", "DOGE", "ADA", "AVAX", "DOT", "LINK",
"MATIC", "UNI", "LTC", "BCH", "NEAR", "APT", "OP", "ARB", "FIL", "ATOM",
"ETC", "IMX", "INJ", "SEI", "SUI", "TIA", "JUP", "WLD", "PEPE", "WIF",
]
# 交易所配置
EXCHANGES = {
"binance": {
"name": "Binance",
"base_url": "https://api.binance.com",
"futures_url": "https://fapi.binance.com",
"endpoints": {
"funding_rate": "/fapi/v1/premiumIndex",
"funding_history": "/fapi/v1/fundingRate",
"spot_ticker": "/api/v3/ticker/bookTicker",
"futures_ticker": "/fapi/v1/ticker/price",
"futures_24h": "/fapi/v1/ticker/24hr",
"spot_price": "/api/v3/ticker/price",
"open_interest": "/fapi/v1/openInterest",
"long_short_ratio": "/futures/data/globalLongShortAccountRatio",
"exchange_info": "/api/v3/exchangeInfo",
},
"symbol_format": lambda base: f"{base}USDT",
"swap_format": lambda base: f"{base}USDT",
"funding_interval_hours": 8,
},
"bybit": {
"name": "Bybit",
"base_url": "https://api.bybit.com",
"futures_url": "https://api.bybit.com",
"endpoints": {
"funding_rate": "/v5/market/tickers",
"funding_history": "/v5/market/funding/history",
"spot_ticker": "/v5/market/tickers",
"futures_ticker": "/v5/market/tickers",
"open_interest": "/v5/market/open-interest",
"long_short_ratio": "/v5/market/account-ratio",
"instruments": "/v5/market/instruments-info",
},
"symbol_format": lambda base: f"{base}USDT",
"swap_format": lambda base: f"{base}USDT",
"funding_interval_hours": 8,
},
"okx": {
"name": "OKX",
"base_url": "https://www.okx.com",
"futures_url": "https://www.okx.com",
"endpoints": {
"funding_rate": "/api/v5/public/funding-rate",
"funding_history": "/api/v5/public/funding-rate-history",
"spot_ticker": "/api/v5/market/ticker",
"futures_ticker": "/api/v5/market/ticker",
"open_interest": "/api/v5/public/open-interest",
"instruments": "/api/v5/public/instruments",
},
"symbol_format": lambda base: f"{base}-USDT",
"swap_format": lambda base: f"{base}-USDT-SWAP",
"funding_interval_hours": 8,
},
"bitget": {
"name": "Bitget",
"base_url": "https://api.bitget.com",
"futures_url": "https://api.bitget.com",
"endpoints": {
"funding_rate": "/api/v2/mix/market/current-fund-rate",
"funding_history": "/api/v2/mix/market/history-fund-rate",
"spot_ticker": "/api/v2/spot/market/tickers",
"futures_ticker": "/api/v2/mix/market/tickers",
"open_interest": "/api/v2/mix/market/open-interest",
"instruments": "/api/v2/spot/public/symbols",
},
"symbol_format": lambda base: f"{base}USDT",
"swap_format": lambda base: f"{base}USDT",
"funding_interval_hours": 8,
},
}
# 请求配置
REQUEST_TIMEOUT = 10 # 秒
RATE_LIMIT_DELAY = 0.2 # 秒，请求间隔
FILE:scripts/fetcher.py
"""统一数据获取层 — 公开 API，无需 key"""
import time
import requests
from typing import Optional
from config import EXCHANGES, REQUEST_TIMEOUT, RATE_LIMIT_DELAY
_last_request_time = 0
def _rate_limit():
"""限频：请求间隔至少 RATE_LIMIT_DELAY 秒"""
global _last_request_time
elapsed = time.time() - _last_request_time
if elapsed < RATE_LIMIT_DELAY:
time.sleep(RATE_LIMIT_DELAY - elapsed)
_last_request_time = time.time()
def _get(url: str, params: Optional[dict] = None) -> Optional[dict]:
"""发起 GET 请求，返回 JSON 或 None"""
_rate_limit()
try:
resp = requests.get(url, params=params, timeout=REQUEST_TIMEOUT)
resp.raise_for_status()
return resp.json()
except Exception as e:
print(f" [WARN] 请求失败 {url}: {e}")
return None
# ====== 资金费率 ======
def fetch_funding_rate_binance(symbol: str) -> Optional[float]:
cfg = EXCHANGES["binance"]
url = cfg["futures_url"] + cfg["endpoints"]["funding_rate"]
data = _get(url, {"symbol": cfg["swap_format"](symbol)})
if data:
return float(data.get("lastFundingRate", 0))
return None
def fetch_funding_rate_bybit(symbol: str) -> Optional[float]:
cfg = EXCHANGES["bybit"]
url = cfg["base_url"] + cfg["endpoints"]["funding_rate"]
data = _get(url, {"category": "linear", "symbol": cfg["swap_format"](symbol)})
if data and data.get("result", {}).get("list"):
return float(data["result"]["list"][0].get("fundingRate", 0))
return None
def fetch_funding_rate_okx(symbol: str) -> Optional[float]:
cfg = EXCHANGES["okx"]
url = cfg["base_url"] + cfg["endpoints"]["funding_rate"]
data = _get(url, {"instId": cfg["swap_format"](symbol)})
if data and data.get("data"):
return float(data["data"][0].get("fundingRate", 0))
return None
def fetch_funding_rate_bitget(symbol: str) -> Optional[float]:
cfg = EXCHANGES["bitget"]
url = cfg["base_url"] + cfg["endpoints"]["funding_rate"]
data = _get(url, {"symbol": cfg["swap_format"](symbol), "productType": "USDT-FUTURES"})
if data and data.get("data"):
return float(data["data"][0].get("fundingRate", 0))
return None
def fetch_all_funding_rates(symbol: str) -> dict:
"""获取某个 symbol 在所有交易所的资金费率"""
fetchers = {
"binance": fetch_funding_rate_binance,
"bybit": fetch_funding_rate_bybit,
"okx": fetch_funding_rate_okx,
"bitget": fetch_funding_rate_bitget,
}
results = {}
for exchange, fetcher in fetchers.items():
rate = fetcher(symbol)
if rate is not None:
results[exchange] = rate
return results
# ====== 现货行情 ======
def fetch_spot_ticker_binance(symbol: str) -> Optional[dict]:
cfg = EXCHANGES["binance"]
url = cfg["base_url"] + cfg["endpoints"]["spot_ticker"]
data = _get(url, {"symbol": cfg["symbol_format"](symbol)})
if data:
return {"bid": float(data["bidPrice"]), "ask": float(data["askPrice"])}
return None
def fetch_spot_ticker_bybit(symbol: str) -> Optional[dict]:
cfg = EXCHANGES["bybit"]
url = cfg["base_url"] + cfg["endpoints"]["spot_ticker"]
data = _get(url, {"category": "spot", "symbol": cfg["symbol_format"](symbol)})
if data and data.get("result", {}).get("list"):
item = data["result"]["list"][0]
return {"bid": float(item["bid1Price"]), "ask": float(item["ask1Price"])}
return None
def fetch_spot_ticker_okx(symbol: str) -> Optional[dict]:
cfg = EXCHANGES["okx"]
url = cfg["base_url"] + cfg["endpoints"]["spot_ticker"]
data = _get(url, {"instId": cfg["symbol_format"](symbol)})
if data and data.get("data"):
item = data["data"][0]
return {"bid": float(item["bidPx"]), "ask": float(item["askPx"])}
return None
def fetch_spot_ticker_bitget(symbol: str) -> Optional[dict]:
cfg = EXCHANGES["bitget"]
url = cfg["base_url"] + cfg["endpoints"]["spot_ticker"]
# Bitget tickers 接口不支持单 symbol 过滤，需要遍历
data = _get(url)
if data and data.get("data"):
sym = cfg["symbol_format"](symbol)
for item in data["data"]:
if item.get("symbol") == sym:
return {"bid": float(item.get("bidPr", 0)), "ask": float(item.get("askPr", 0))}
return None
def fetch_all_spot_tickers(symbol: str) -> dict:
"""获取某个 symbol 在所有交易所的现货 bid/ask"""
fetchers = {
"binance": fetch_spot_ticker_binance,
"bybit": fetch_spot_ticker_bybit,
"okx": fetch_spot_ticker_okx,
"bitget": fetch_spot_ticker_bitget,
}
results = {}
for exchange, fetcher in fetchers.items():
ticker = fetcher(symbol)
if ticker and ticker["bid"] > 0 and ticker["ask"] > 0:
results[exchange] = ticker
return results
# ====== 合约行情 ======
def fetch_futures_price_binance(symbol: str) -> Optional[float]:
cfg = EXCHANGES["binance"]
url = cfg["futures_url"] + cfg["endpoints"]["futures_ticker"]
data = _get(url, {"symbol": cfg["swap_format"](symbol)})
if data:
return float(data["price"])
return None
def fetch_futures_price_bybit(symbol: str) -> Optional[float]:
cfg = EXCHANGES["bybit"]
url = cfg["base_url"] + cfg["endpoints"]["futures_ticker"]
data = _get(url, {"category": "linear", "symbol": cfg["swap_format"](symbol)})
if data and data.get("result", {}).get("list"):
return float(data["result"]["list"][0].get("lastPrice", 0))
return None
def fetch_futures_price_okx(symbol: str) -> Optional[float]:
cfg = EXCHANGES["okx"]
url = cfg["base_url"] + cfg["endpoints"]["futures_ticker"]
data = _get(url, {"instId": cfg["swap_format"](symbol)})
if data and data.get("data"):
return float(data["data"][0].get("last", 0))
return None
def fetch_futures_price_bitget(symbol: str) -> Optional[float]:
cfg = EXCHANGES["bitget"]
url = cfg["base_url"] + cfg["endpoints"]["futures_ticker"]
# Bitget tickers 接口不支持单 symbol 过滤，需要遍历
data = _get(url, {"productType": "USDT-FUTURES"})
if data and data.get("data"):
sym = cfg["swap_format"](symbol)
for item in data["data"]:
if item.get("symbol") == sym:
return float(item.get("lastPr", 0))
return None
def fetch_all_futures_prices(symbol: str) -> dict:
"""获取某个 symbol 在所有交易所的合约价格"""
fetchers = {
"binance": fetch_futures_price_binance,
"bybit": fetch_futures_price_bybit,
"okx": fetch_futures_price_okx,
"bitget": fetch_futures_price_bitget,
}
results = {}
for exchange, fetcher in fetchers.items():
price = fetcher(symbol)
if price and price > 0:
results[exchange] = price
return results
# ====== 现货价格（简单版，用于期现基差） ======
def fetch_spot_price_binance(symbol: str) -> Optional[float]:
cfg = EXCHANGES["binance"]
url = cfg["base_url"] + cfg["endpoints"]["spot_price"]
data = _get(url, {"symbol": cfg["symbol_format"](symbol)})
if data:
return float(data["price"])
return None
def fetch_spot_price_bybit(symbol: str) -> Optional[float]:
ticker = fetch_spot_ticker_bybit(symbol)
if ticker:
return (ticker["bid"] + ticker["ask"]) / 2
return None
def fetch_spot_price_okx(symbol: str) -> Optional[float]:
ticker = fetch_spot_ticker_okx(symbol)
if ticker:
return (ticker["bid"] + ticker["ask"]) / 2
return None
def fetch_spot_price_bitget(symbol: str) -> Optional[float]:
ticker = fetch_spot_ticker_bitget(symbol)
if ticker:
return (ticker["bid"] + ticker["ask"]) / 2
return None
def fetch_all_spot_prices(symbol: str) -> dict:
"""获取某个 symbol 在所有交易所的现货中间价"""
fetchers = {
"binance": fetch_spot_price_binance,
"bybit": fetch_spot_price_bybit,
"okx": fetch_spot_price_okx,
"bitget": fetch_spot_price_bitget,
}
results = {}
for exchange, fetcher in fetchers.items():
price = fetcher(symbol)
if price and price > 0:
results[exchange] = price
return results
# ====== 持仓量 (Open Interest) ======
def fetch_oi_binance(symbol: str) -> Optional[float]:
cfg = EXCHANGES["binance"]
url = cfg["futures_url"] + cfg["endpoints"]["open_interest"]
data = _get(url, {"symbol": cfg["swap_format"](symbol)})
if data:
return float(data.get("openInterest", 0))
return None
def fetch_oi_bybit(symbol: str) -> Optional[float]:
cfg = EXCHANGES["bybit"]
url = cfg["base_url"] + cfg["endpoints"]["open_interest"]
data = _get(url, {"category": "linear", "symbol": cfg["swap_format"](symbol), "intervalTime": "5min"})
if data and data.get("result", {}).get("list"):
return float(data["result"]["list"][0].get("openInterest", 0))
return None
def fetch_oi_okx(symbol: str) -> Optional[float]:
cfg = EXCHANGES["okx"]
url = cfg["base_url"] + cfg["endpoints"]["open_interest"]
data = _get(url, {"instType": "SWAP", "instId": cfg["swap_format"](symbol)})
if data and data.get("data"):
return float(data["data"][0].get("oi", 0))
return None
def fetch_oi_bitget(symbol: str) -> Optional[float]:
cfg = EXCHANGES["bitget"]
url = cfg["base_url"] + cfg["endpoints"]["open_interest"]
data = _get(url, {"productType": "USDT-FUTURES", "symbol": cfg["swap_format"](symbol)})
if data and data.get("data"):
oi_list = data["data"].get("openInterestList")
if oi_list and len(oi_list) > 0:
return float(oi_list[0].get("size", 0))
return None
def fetch_all_open_interest(symbol: str) -> dict:
"""获取某个 symbol 在所有交易所的持仓量"""
fetchers = {
"binance": fetch_oi_binance,
"bybit": fetch_oi_bybit,
"okx": fetch_oi_okx,
"bitget": fetch_oi_bitget,
}
results = {}
for exchange, fetcher in fetchers.items():
oi = fetcher(symbol)
if oi and oi > 0:
results[exchange] = oi
return results
# ====== 24h 行情（价格变化 + 成交量） ======
def fetch_24h_ticker_binance(symbol: str) -> Optional[dict]:
cfg = EXCHANGES["binance"]
url = cfg["futures_url"] + cfg["endpoints"]["futures_24h"]
data = _get(url, {"symbol": cfg["swap_format"](symbol)})
if data:
return {
"exchange_name": cfg["name"],
"last_price": float(data.get("lastPrice", 0)),
"change_pct": float(data.get("priceChangePercent", 0)),
"volume_usdt": float(data.get("quoteVolume", 0)),
}
return None
def fetch_24h_ticker_bybit(symbol: str) -> Optional[dict]:
cfg = EXCHANGES["bybit"]
url = cfg["base_url"] + cfg["endpoints"]["futures_ticker"]
data = _get(url, {"category": "linear", "symbol": cfg["swap_format"](symbol)})
if data and data.get("result", {}).get("list"):
item = data["result"]["list"][0]
price = float(item.get("lastPrice", 0))
prev = float(item.get("prevPrice24h", 0))
change_pct = ((price - prev) / prev * 100) if prev > 0 else 0
return {
"exchange_name": cfg["name"],
"last_price": price,
"change_pct": change_pct,
"volume_usdt": float(item.get("turnover24h", 0)),
}
return None
def fetch_24h_ticker_okx(symbol: str) -> Optional[dict]:
cfg = EXCHANGES["okx"]
url = cfg["base_url"] + cfg["endpoints"]["futures_ticker"]
data = _get(url, {"instId": cfg["swap_format"](symbol)})
if data and data.get("data"):
item = data["data"][0]
last = float(item.get("last", 0))
open24h = float(item.get("open24h", 0))
change_pct = ((last - open24h) / open24h * 100) if open24h > 0 else 0
return {
"exchange_name": cfg["name"],
"last_price": last,
"change_pct": change_pct,
"volume_usdt": float(item.get("volCcy24h", 0)),
}
return None
def fetch_24h_ticker_bitget(symbol: str) -> Optional[dict]:
cfg = EXCHANGES["bitget"]
url = cfg["base_url"] + cfg["endpoints"]["futures_ticker"]
data = _get(url, {"productType": "USDT-FUTURES"})
if data and data.get("data"):
sym = cfg["swap_format"](symbol)
for item in data["data"]:
if item.get("symbol") == sym:
return {
"exchange_name": cfg["name"],
"last_price": float(item.get("lastPr", 0)),
"change_pct": float(item.get("change24h", 0)) * 100, # Bitget 返回小数形式
"volume_usdt": float(item.get("quoteVolume", 0)),
}
return None
def fetch_all_24h_tickers(symbol: str) -> dict:
"""获取某个 symbol 在所有交易所的 24h 行情"""
fetchers = {
"binance": fetch_24h_ticker_binance,
"bybit": fetch_24h_ticker_bybit,
"okx": fetch_24h_ticker_okx,
"bitget": fetch_24h_ticker_bitget,
}
results = {}
for exchange, fetcher in fetchers.items():
data = fetcher(symbol)
if data and data["last_price"] > 0:
results[exchange] = data
return results
# ====== 资金费率历史 ======
def fetch_funding_history(exchange: str, symbol: str, limit: int = 20) -> list:
"""获取某个交易所某个 symbol 的历史资金费率，返回费率列表（最新在前）"""
cfg = EXCHANGES.get(exchange)
if not cfg or "funding_history" not in cfg["endpoints"]:
return []
url = (cfg.get("futures_url") or cfg["base_url"]) + cfg["endpoints"]["funding_history"]
if exchange == "binance":
data = _get(url, {"symbol": cfg["swap_format"](symbol), "limit": limit})
if data and isinstance(data, list):
return [float(r["fundingRate"]) for r in reversed(data)]
elif exchange == "bybit":
data = _get(url, {"category": "linear", "symbol": cfg["swap_format"](symbol), "limit": limit})
if data and data.get("result", {}).get("list"):
return [float(r["fundingRate"]) for r in data["result"]["list"]]
elif exchange == "okx":
data = _get(url, {"instId": cfg["swap_format"](symbol), "limit": limit})
if data and data.get("data"):
return [float(r["fundingRate"]) for r in data["data"]]
elif exchange == "bitget":
data = _get(url, {"symbol": cfg["swap_format"](symbol), "productType": "USDT-FUTURES", "pageSize": limit})
if data and data.get("data"):
return [float(r["fundingRate"]) for r in data["data"]]
return []
# ====== 多空比 ======
def fetch_long_short_ratio(exchange: str, symbol: str) -> Optional[dict]:
"""获取多空比，返回 {long_pct, short_pct}"""
cfg = EXCHANGES.get(exchange)
if not cfg or "long_short_ratio" not in cfg["endpoints"]:
return None
url = cfg.get("base_url", cfg.get("futures_url", "")) + cfg["endpoints"]["long_short_ratio"]
if exchange == "binance":
# Binance 的多空比端点在 fapi 域名下
url = cfg["futures_url"] + cfg["endpoints"]["long_short_ratio"]
data = _get(url, {"symbol": cfg["swap_format"](symbol), "period": "5m", "limit": 1})
if data and isinstance(data, list) and len(data) > 0:
ratio = float(data[0].get("longShortRatio", 1))
long_pct = ratio / (1 + ratio) * 100
short_pct = 100 - long_pct
return {"long_pct": long_pct, "short_pct": short_pct}
elif exchange == "bybit":
data = _get(url, {"category": "linear", "symbol": cfg["swap_format"](symbol), "period": "1h", "limit": 1})
if data and data.get("result", {}).get("list"):
item = data["result"]["list"][0]
buy_ratio = float(item.get("buyRatio", 0.5))
sell_ratio = float(item.get("sellRatio", 0.5))
total = buy_ratio + sell_ratio
if total > 0:
return {"long_pct": buy_ratio / total * 100, "short_pct": sell_ratio / total * 100}
return None
# ====== 交易对列表 ======
def fetch_exchange_symbols(exchange: str) -> Optional[set]:
"""获取某交易所所有 USDT 交易对 symbol（如 BTCUSDT）"""
cfg = EXCHANGES.get(exchange)
if not cfg:
return None
if exchange == "binance":
url = cfg["base_url"] + cfg["endpoints"]["exchange_info"]
data = _get(url)
if data and data.get("symbols"):
return {s["symbol"] for s in data["symbols"] if s.get("quoteAsset") == "USDT" and s.get("status") == "TRADING"}
elif exchange == "bybit":
url = cfg["base_url"] + cfg["endpoints"]["instruments"]
data = _get(url, {"category": "spot"})
if data and data.get("result", {}).get("list"):
return {s["symbol"] for s in data["result"]["list"] if s["symbol"].endswith("USDT") and s.get("status") == "Trading"}
elif exchange == "okx":
url = cfg["base_url"] + cfg["endpoints"]["instruments"]
data = _get(url, {"instType": "SPOT"})
if data and data.get("data"):
return {s["instId"].replace("-", "") for s in data["data"] if s["instId"].endswith("-USDT") and s.get("state") == "live"}
elif exchange == "bitget":
url = cfg["base_url"] + cfg["endpoints"]["instruments"]
data = _get(url)
if data and data.get("data"):
return {s["symbol"] for s in data["data"] if s["symbol"].endswith("USDT") and s.get("status") == "online"}
return None
FILE:scripts/formatter.py
"""输出格式化 — table / json / markdown"""
import json
from tabulate import tabulate
def format_table(rows: list, headers: list) -> str:
"""格式化为终端表格"""
return tabulate(rows, headers=headers, tablefmt="simple", floatfmt=".4f")
def format_markdown(rows: list, headers: list) -> str:
"""格式化为 Markdown 表格"""
return tabulate(rows, headers=headers, tablefmt="github", floatfmt=".4f")
def format_json(rows: list, headers: list) -> str:
"""格式化为 JSON"""
data = [dict(zip(headers, row)) for row in rows]
return json.dumps(data, indent=2, ensure_ascii=False)
def format_output(rows: list, headers: list, fmt: str = "table") -> str:
"""统一格式化入口"""
if not rows:
return "No opportunities found."
formatters = {
"table": format_table,
"markdown": format_markdown,
"json": format_json,
}
formatter = formatters.get(fmt, format_table)
return formatter(rows, headers)
def risk_level(apy: float, symbol: str) -> str:
"""根据年化和币种评估风险等级
HIGH: APY > 50% 或 (非主流币 且 APY > 20%)
MEDIUM: APY 10-50% 或 非主流币
LOW: 主流币 且 APY < 10%
"""
major_coins = {"BTC", "ETH", "BNB", "SOL", "XRP"}
is_major = symbol in major_coins
if apy > 50:
return "HIGH"
if not is_major and apy > 20:
return "HIGH"
if apy > 10 or not is_major:
return "MEDIUM"
return "LOW"
FILE:scripts/funding_arb.py
"""资金费率套利扫描"""
from config import TOP_SYMBOLS, EXCHANGES
from fetcher import fetch_all_funding_rates
from formatter import format_output, risk_level
def scan_funding_arbitrage(symbols: list = None, min_apy: float = 0) -> tuple:
"""
扫描资金费率套利机会
返回 (rows, headers)
"""
if symbols is None:
symbols = TOP_SYMBOLS
headers = ["Symbol", "Long (低费率)", "Short (高费率)", "Rate Diff", "Est. APY%", "Risk", "Window"]
rows = []
print(f"[Funding Rate] 扫描 {len(symbols)} 个交易对...")
for symbol in symbols:
rates = fetch_all_funding_rates(symbol)
if len(rates) < 2:
continue
# 找最低和最高费率
min_ex = min(rates, key=rates.get)
max_ex = max(rates, key=rates.get)
diff = rates[max_ex] - rates[min_ex]
if diff <= 0:
continue
# 年化计算：rate_diff * (365 * 24 / interval_hours) * 100
interval = EXCHANGES[min_ex]["funding_interval_hours"]
apy = diff * (365 * 24 / interval) * 100
if apy < min_apy:
continue
risk = risk_level(apy, symbol)
rows.append([
f"{symbol}USDT",
f"{EXCHANGES[min_ex]['name']} {rates[min_ex]*100:.4f}%",
f"{EXCHANGES[max_ex]['name']} {rates[max_ex]*100:.4f}%",
f"{diff*100:.4f}%",
f"{apy:.1f}%",
risk,
f"~{interval}h",
])
# 按年化降序排列
rows.sort(key=lambda r: float(r[4].rstrip('%')), reverse=True)
return rows, headers
def main():
import argparse
parser = argparse.ArgumentParser(description="Funding Rate Arbitrage Scanner")
parser.add_argument("--min-apy", type=float, default=0, help="最低年化过滤")
parser.add_argument("--format", choices=["table", "markdown", "json"], default="table")
args = parser.parse_args()
rows, headers = scan_funding_arbitrage(min_apy=args.min_apy)
print(f"\n{'='*80}")
print(" Funding Rate Arbitrage Opportunities")
print(f"{'='*80}")
print(format_output(rows, headers, args.format))
print(f"\nFound {len(rows)} opportunities")
if __name__ == "__main__":
main()
FILE:scripts/funding_extreme.py
"""资金费率异常警报 — 费率 > ±0.1% 的极端情况"""
from config import TOP_SYMBOLS, EXCHANGES
from fetcher import fetch_all_funding_rates
from formatter import format_output
# 正常费率约 ±0.01%，超过 ±0.1% 算极端
EXTREME_THRESHOLD = 0.001 # 0.1%
def scan_funding_extreme(symbols: list = None) -> tuple:
"""扫描资金费率极端值"""
if symbols is None:
symbols = TOP_SYMBOLS
headers = ["Symbol", "Exchange", "Rate", "Multiple", "Direction", "Signal"]
rows = []
print(f"[Funding Extreme] 扫描 {len(symbols)} 个交易对...")
for symbol in symbols:
rates = fetch_all_funding_rates(symbol)
for exchange, rate in rates.items():
if abs(rate) < EXTREME_THRESHOLD:
continue
# 正常费率 0.01%，计算是正常的多少倍
multiple = abs(rate) / 0.0001
direction = "LONG CROWDED" if rate > 0 else "SHORT CROWDED"
# 极端费率意味着反方向可能有机会
signal = "Potential SHORT squeeze" if rate < 0 else "Potential LONG squeeze"
rows.append([
f"{symbol}USDT",
EXCHANGES[exchange]["name"],
f"{rate*100:.4f}%",
f"{multiple:.1f}x",
direction,
signal,
])
rows.sort(key=lambda r: float(r[3].rstrip('x')), reverse=True)
return rows, headers
def main():
import argparse
parser = argparse.ArgumentParser(description="Funding Rate Extreme Alert")
parser.add_argument("--format", choices=["table", "markdown", "json"], default="table")
args = parser.parse_args()
rows, headers = scan_funding_extreme()
print(f"\n{'='*80}")
print(" Funding Rate Extreme Alert")
print(f"{'='*80}")
print(format_output(rows, headers, args.format))
print(f"\nFound {len(rows)} extreme rates")
if __name__ == "__main__":
main()
FILE:scripts/funding_history.py
"""资金费率历史趋势 — 找持续同方向费率的稳定套利机会"""
from config import TOP_SYMBOLS, EXCHANGES
from fetcher import fetch_funding_history
from formatter import format_output
def scan_funding_history(symbols: list = None, min_streak: int = 5) -> tuple:
"""扫描资金费率历史趋势，找连续同方向的币种"""
if symbols is None:
symbols = TOP_SYMBOLS[:30] # 历史费率查询较慢，默认 Top 30
headers = ["Symbol", "Exchange", "Streak", "Direction", "Avg Rate", "Est. APY%", "Stability"]
rows = []
print(f"[Funding History] 扫描 {len(symbols)} 个交易对...")
for symbol in symbols:
for exchange in ["binance", "bybit", "okx"]:
rates = fetch_funding_history(exchange, symbol, limit=20)
if len(rates) < min_streak:
continue
# 计算连续同方向的长度
streak = 1
direction = "POSITIVE" if rates[0] > 0 else "NEGATIVE"
for i in range(1, len(rates)):
if (rates[i] > 0) == (rates[0] > 0):
streak += 1
else:
break
if streak < min_streak:
continue
avg_rate = sum(rates[:streak]) / streak
interval = EXCHANGES[exchange]["funding_interval_hours"]
apy = avg_rate * (365 * 24 / interval) * 100
if streak >= 15:
stability = "VERY STABLE"
elif streak >= 10:
stability = "STABLE"
else:
stability = "EMERGING"
rows.append([
f"{symbol}USDT",
EXCHANGES[exchange]["name"],
f"{streak} periods",
direction,
f"{avg_rate*100:.4f}%",
f"{apy:.1f}%",
stability,
])
rows.sort(key=lambda r: int(r[2].split()[0]), reverse=True)
return rows, headers
def main():
import argparse
parser = argparse.ArgumentParser(description="Funding Rate History Trend")
parser.add_argument("--format", choices=["table", "markdown", "json"], default="table")
parser.add_argument("--min-streak", type=int, default=5, help="最小连续期数")
args = parser.parse_args()
rows, headers = scan_funding_history(min_streak=args.min_streak)
print(f"\n{'='*80}")
print(" Funding Rate History Trend")
print(f"{'='*80}")
print(format_output(rows, headers, args.format))
print(f"\nFound {len(rows)} trending symbols")
if __name__ == "__main__":
main()
FILE:scripts/futures_spread.py
"""跨所合约价差扫描 — 合约端的搬砖机会"""
from itertools import combinations
from config import TOP_SYMBOLS, EXCHANGES
from fetcher import fetch_all_futures_prices
from formatter import format_output, risk_level
def scan_futures_spread(symbols: list = None, min_spread_pct: float = 0.03) -> tuple:
"""扫描跨所合约价差"""
if symbols is None:
symbols = TOP_SYMBOLS
headers = ["Symbol", "Buy From", "Sell To", "Low Price", "High Price", "Spread%", "Risk"]
rows = []
print(f"[Futures Spread] 扫描 {len(symbols)} 个交易对...")
for symbol in symbols:
prices = fetch_all_futures_prices(symbol)
if len(prices) < 2:
continue
for ex_a, ex_b in combinations(prices.keys(), 2):
pa, pb = prices[ex_a], prices[ex_b]
if pa <= 0 or pb <= 0:
continue
low_ex, high_ex = (ex_a, ex_b) if pa < pb else (ex_b, ex_a)
low_p, high_p = min(pa, pb), max(pa, pb)
spread_pct = (high_p - low_p) / low_p * 100
if spread_pct < min_spread_pct:
continue
risk = risk_level(spread_pct * 365, symbol)
rows.append([
f"{symbol}USDT",
f"{EXCHANGES[low_ex]['name']} .4f",
f"{EXCHANGES[high_ex]['name']} .4f",
f".4f",
f".4f",
f"{spread_pct:.4f}%",
risk,
])
rows.sort(key=lambda r: float(r[5].rstrip('%')), reverse=True)
return rows, headers
def main():
import argparse
parser = argparse.ArgumentParser(description="Cross-Exchange Futures Spread Scanner")
parser.add_argument("--format", choices=["table", "markdown", "json"], default="table")
args = parser.parse_args()
rows, headers = scan_futures_spread()
print(f"\n{'='*80}")
print(" Cross-Exchange Futures Spread")
print(f"{'='*80}")
print(format_output(rows, headers, args.format))
print(f"\nFound {len(rows)} opportunities")
if __name__ == "__main__":
main()
FILE:scripts/long_short_ratio.py
"""多空比扫描 — 散户情绪一边倒时的反向信号"""
from config import TOP_SYMBOLS
from fetcher import fetch_long_short_ratio
from formatter import format_output
def scan_long_short_ratio(symbols: list = None, min_ratio_pct: float = 60) -> tuple:
"""扫描多空比极端值"""
if symbols is None:
symbols = TOP_SYMBOLS[:30] # 多空比查询较慢，默认 Top 30
headers = ["Symbol", "Exchange", "Long%", "Short%", "Ratio", "Signal"]
rows = []
print(f"[Long/Short Ratio] 扫描 {len(symbols)} 个交易对...")
for symbol in symbols:
# Binance 和 Bybit 有公开多空比端点
for exchange in ["binance", "bybit"]:
data = fetch_long_short_ratio(exchange, symbol)
if data is None:
continue
long_pct = data["long_pct"]
short_pct = data["short_pct"]
# 只标记极端情况
if max(long_pct, short_pct) < min_ratio_pct:
continue
ratio = long_pct / short_pct if short_pct > 0 else float('inf')
if long_pct > 75:
signal = "EXTREME LONG - reversal risk"
elif long_pct > 65:
signal = "LONG HEAVY"
elif short_pct > 75:
signal = "EXTREME SHORT - squeeze risk"
elif short_pct > 65:
signal = "SHORT HEAVY"
else:
signal = "MODERATE"
from config import EXCHANGES
rows.append([
f"{symbol}USDT",
EXCHANGES[exchange]["name"],
f"{long_pct:.1f}%",
f"{short_pct:.1f}%",
f"{ratio:.2f}",
signal,
])
rows.sort(key=lambda r: max(float(r[2].rstrip('%')), float(r[3].rstrip('%'))), reverse=True)
return rows, headers
def main():
import argparse
parser = argparse.ArgumentParser(description="Long/Short Ratio Scanner")
parser.add_argument("--format", choices=["table", "markdown", "json"], default="table")
args = parser.parse_args()
rows, headers = scan_long_short_ratio()
print(f"\n{'='*80}")
print(" Long/Short Ratio")
print(f"{'='*80}")
print(format_output(rows, headers, args.format))
print(f"\nFound {len(rows)} extreme ratios")
if __name__ == "__main__":
main()
FILE:scripts/new_listing.py
"""新币上线检测 — A 所有 B 所没有，找溢价窗口"""
from fetcher import fetch_exchange_symbols
from formatter import format_output
def scan_new_listing() -> tuple:
"""比较各交易所交易对列表，找出独有的币种"""
headers = ["Symbol", "Available On", "Missing From", "Exclusivity", "Signal"]
rows = []
print("[New Listing] 获取各交易所交易对列表...")
all_symbols = {}
for exchange in ["binance", "bybit", "okx", "bitget"]:
symbols = fetch_exchange_symbols(exchange)
if symbols:
all_symbols[exchange] = symbols
print(f" {exchange}: {len(symbols)} pairs")
if len(all_symbols) < 2:
return rows, headers
# 合并所有 symbol
universe = set()
for syms in all_symbols.values():
universe.update(syms)
from config import EXCHANGES
for symbol in sorted(universe):
present = [ex for ex in all_symbols if symbol in all_symbols[ex]]
missing = [ex for ex in all_symbols if symbol not in all_symbols[ex]]
# 只关注在 1-2 个交易所独有的
if len(present) >= 3 or len(missing) == 0:
continue
present_names = ", ".join(EXCHANGES[ex]["name"] for ex in present)
missing_names = ", ".join(EXCHANGES[ex]["name"] for ex in missing)
exclusivity = f"{len(present)}/{len(all_symbols)}"
if len(present) == 1:
signal = "EXCLUSIVE - high premium potential"
else:
signal = "LIMITED - moderate premium"
rows.append([
symbol,
present_names,
missing_names,
exclusivity,
signal,
])
# 按独有度排序（越少交易所有越前）
rows.sort(key=lambda r: r[3])
return rows, headers
def main():
import argparse
parser = argparse.ArgumentParser(description="New Listing Detection")
parser.add_argument("--format", choices=["table", "markdown", "json"], default="table")
args = parser.parse_args()
rows, headers = scan_new_listing()
print(f"\n{'='*80}")
print(" New Listing Detection")
print(f"{'='*80}")
print(format_output(rows, headers, args.format))
print(f"\nFound {len(rows)} exclusive/limited listings")
if __name__ == "__main__":
main()
FILE:scripts/open_interest.py
"""持仓量监控 — 跨所 OI 对比，发现异常堆积"""
from config import TOP_SYMBOLS, EXCHANGES
from fetcher import fetch_all_open_interest
from formatter import format_output
def scan_open_interest(symbols: list = None, top_n: int = 30) -> tuple:
"""扫描持仓量分布，找出 OI 集中度异常的币种"""
if symbols is None:
symbols = TOP_SYMBOLS[:top_n] # OI 扫描默认 Top 30，节省时间
headers = ["Symbol", "Total OI (USDT)", "Top Exchange", "Share%", "Status"]
rows = []
print(f"[Open Interest] 扫描 {len(symbols)} 个交易对...")
for symbol in symbols:
oi_data = fetch_all_open_interest(symbol)
if len(oi_data) < 2:
continue
total_oi = sum(oi_data.values())
if total_oi <= 0:
continue
top_ex = max(oi_data, key=oi_data.get)
top_share = oi_data[top_ex] / total_oi * 100
if top_share > 60:
status = "CONCENTRATED"
elif top_share > 45:
status = "MODERATE"
else:
status = "BALANCED"
rows.append([
f"{symbol}USDT",
f",.0f",
f"{EXCHANGES[top_ex]['name']} ({top_share:.1f}%)",
f"{top_share:.1f}%",
status,
])
rows.sort(key=lambda r: float(r[3].rstrip('%')), reverse=True)
return rows, headers
def main():
import argparse
parser = argparse.ArgumentParser(description="Open Interest Monitor")
parser.add_argument("--format", choices=["table", "markdown", "json"], default="table")
parser.add_argument("--top", type=int, default=30, help="扫描前 N 个币种")
args = parser.parse_args()
rows, headers = scan_open_interest(top_n=args.top)
print(f"\n{'='*80}")
print(" Open Interest Monitor")
print(f"{'='*80}")
print(format_output(rows, headers, args.format))
print(f"\nMonitoring {len(rows)} symbols")
if __name__ == "__main__":
main()
FILE:scripts/price_movers.py
"""24h 涨跌幅排行 — 暴涨暴跌币种检测"""
from config import TOP_SYMBOLS
from fetcher import fetch_all_24h_tickers
from formatter import format_output
def scan_price_movers(symbols: list = None, top_n: int = 20) -> tuple:
"""扫描 24h 涨跌幅最大的币种"""
if symbols is None:
symbols = TOP_SYMBOLS
headers = ["Symbol", "Exchange", "Price", "24h Change%", "24h Volume (USDT)", "Direction"]
all_rows = []
print(f"[Price Movers] 扫描 {len(symbols)} 个交易对...")
for symbol in symbols:
tickers = fetch_all_24h_tickers(symbol)
for exchange, data in tickers.items():
change_pct = data.get("change_pct", 0)
volume = data.get("volume_usdt", 0)
price = data.get("last_price", 0)
if price <= 0:
continue
direction = "PUMP" if change_pct > 0 else "DUMP"
all_rows.append([
f"{symbol}USDT",
data.get("exchange_name", exchange),
f".4f",
f"{change_pct:+.2f}%",
f",.0f",
direction,
abs(change_pct), # 用于排序，不输出
])
# 按绝对变化排序，取 top_n
all_rows.sort(key=lambda r: r[6], reverse=True)
# 去掉排序辅助列，去重（同一个币取变化最大的交易所）
seen = set()
rows = []
for r in all_rows:
sym = r[0]
if sym not in seen:
seen.add(sym)
rows.append(r[:6])
if len(rows) >= top_n:
break
return rows, headers
def main():
import argparse
parser = argparse.ArgumentParser(description="24h Price Movers")
parser.add_argument("--format", choices=["table", "markdown", "json"], default="table")
parser.add_argument("--top", type=int, default=20, help="显示前 N 个")
args = parser.parse_args()
rows, headers = scan_price_movers(top_n=args.top)
print(f"\n{'='*80}")
print(" 24h Price Movers (Gainers & Losers)")
print(f"{'='*80}")
print(format_output(rows, headers, args.format))
print(f"\nTop {len(rows)} movers")
if __name__ == "__main__":
main()
FILE:scripts/requirements.txt
requests>=2.28.0
tabulate>=0.9.0
FILE:scripts/scanner.py
"""ArbiScan CLI 统一入口 — 12 种扫描"""
import argparse
import sys
import time
from datetime import datetime, timezone
from funding_arb import scan_funding_arbitrage
from basis_arb import scan_basis_arbitrage
from spot_spread import scan_spot_spread
from futures_spread import scan_futures_spread
from stablecoin_depeg import scan_stablecoin_depeg
from open_interest import scan_open_interest
from funding_extreme import scan_funding_extreme
from price_movers import scan_price_movers
from volume_anomaly import scan_volume_anomaly
from funding_history import scan_funding_history
from long_short_ratio import scan_long_short_ratio
from new_listing import scan_new_listing
from formatter import format_output
BANNER = """
_ _ _ ____
/ \\ _ __ | |__ (_) ___| ___ __ _ _ __
/ _ \\ | '__|| '_ \\| \\___ \\ / __/ _` | '_ \\
/ ___ \\| | | |_) | |___) | (_| (_| | | | |
/_/ \\_\\_| |_.__/|_|____/ \\___\\__,_|_| |_|
Cross-Exchange Crypto Scanner & Monitor v0.2.0
12 scan types. No trading. No API keys needed.
"""
# 扫描器注册表
SCANNERS = {
# 套利类
"funding": ("Funding Rate Arbitrage", scan_funding_arbitrage),
"basis": ("Basis Arbitrage (Spot vs Futures)", scan_basis_arbitrage),
"spread": ("Cross-Exchange Spot Spread", scan_spot_spread),
"futures_spread": ("Cross-Exchange Futures Spread", scan_futures_spread),
# 监控类
"depeg": ("Stablecoin Depeg Monitor", scan_stablecoin_depeg),
"open_interest": ("Open Interest Monitor", scan_open_interest),
"funding_extreme": ("Funding Rate Extreme Alert", scan_funding_extreme),
"price_movers": ("24h Price Movers", scan_price_movers),
"volume_anomaly": ("Volume Anomaly Detection", scan_volume_anomaly),
# 信号类
"funding_history": ("Funding Rate History Trend", scan_funding_history),
"long_short": ("Long/Short Ratio", scan_long_short_ratio),
"new_listing": ("New Listing Detection", scan_new_listing),
}
SCAN_GROUPS = {
"arbitrage": ["funding", "basis", "spread", "futures_spread"],
"monitor": ["depeg", "open_interest", "funding_extreme", "price_movers", "volume_anomaly"],
"signals": ["funding_history", "long_short", "new_listing"],
}
def run_scan(scan_type: str, fmt: str, min_apy: float):
"""运行指定类型的扫描"""
if scan_type == "all":
types_to_run = list(SCANNERS.keys())
elif scan_type in SCAN_GROUPS:
types_to_run = SCAN_GROUPS[scan_type]
elif scan_type in SCANNERS:
types_to_run = [scan_type]
else:
print(f"Unknown scan type: {scan_type}")
print(f"Available: {', '.join(SCANNERS.keys())}")
print(f"Groups: {', '.join(SCAN_GROUPS.keys())}, all")
sys.exit(1)
total_opps = 0
for stype in types_to_run:
title, scanner = SCANNERS[stype]
print(f"\n{'='*80}")
print(f" {title}")
print(f"{'='*80}")
if stype == "funding":
rows, headers = scanner(min_apy=min_apy)
else:
rows, headers = scanner()
print(format_output(rows, headers, fmt))
print(f" -> {len(rows)} results found\n")
total_opps += len(rows)
return total_opps
def main():
all_types = list(SCANNERS.keys()) + list(SCAN_GROUPS.keys())
parser = argparse.ArgumentParser(
description="ArbiScan - Cross-Exchange Crypto Scanner & Monitor",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Scan Types:
Arbitrage: funding, basis, spread, futures_spread
Monitor: depeg, open_interest, funding_extreme, price_movers, volume_anomaly
Signals: funding_history, long_short, new_listing
Groups:
arbitrage - run all arbitrage scans
monitor - run all monitoring scans
signals - run all signal scans
all - run everything
Examples:
python scanner.py --all
python scanner.py --type arbitrage
python scanner.py --type funding --min-apy 10
python scanner.py --type price_movers --format json
"""
)
parser.add_argument("--all", action="store_true", help="运行所有扫描类型")
parser.add_argument("--type", choices=all_types, help="指定扫描类型或分组")
parser.add_argument("--format", choices=["table", "markdown", "json"], default="table", help="输出格式")
parser.add_argument("--min-apy", type=float, default=0, help="最低年化过滤（仅 funding 类型）")
args = parser.parse_args()
if not args.all and not args.type:
parser.print_help()
sys.exit(0)
print(BANNER)
print(f" Scan started at {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')}")
scan_type = "all" if args.all else args.type
start = time.time()
total = run_scan(scan_type, args.format, args.min_apy)
elapsed = time.time() - start
print(f"{'='*80}")
print(f" Scan complete. {total} total results in {elapsed:.1f}s")
print(f"{'='*80}")
if __name__ == "__main__":
main()
FILE:scripts/spot_spread.py
"""跨所现货价差扫描 — best_bid vs best_ask 跨交易所比较"""
from itertools import combinations
from config import TOP_SYMBOLS, EXCHANGES
from fetcher import fetch_all_spot_tickers
from formatter import format_output, risk_level
def scan_spot_spread(symbols: list = None, min_spread_pct: float = 0.02) -> tuple:
"""
扫描跨所现货价差套利机会
返回 (rows, headers)
"""
if symbols is None:
symbols = TOP_SYMBOLS
headers = ["Symbol", "Buy From", "Sell To", "Buy Ask", "Sell Bid", "Spread%", "Risk"]
rows = []
print(f"[Spot Spread] 扫描 {len(symbols)} 个交易对...")
for symbol in symbols:
tickers = fetch_all_spot_tickers(symbol)
if len(tickers) < 2:
continue
# 两两比较
for ex_a, ex_b in combinations(tickers.keys(), 2):
a, b = tickers[ex_a], tickers[ex_b]
# 方向1: 在 A 买（ask），在 B 卖（bid）
if b["bid"] > a["ask"] and a["ask"] > 0:
spread_pct = (b["bid"] - a["ask"]) / a["ask"] * 100
if spread_pct >= min_spread_pct:
risk = risk_level(spread_pct * 365, symbol)
rows.append([
f"{symbol}USDT",
f"{EXCHANGES[ex_a]['name']} .4f",
f"{EXCHANGES[ex_b]['name']} .4f",
f".4f",
f".4f",
f"{spread_pct:.4f}%",
risk,
])
# 方向2: 在 B 买（ask），在 A 卖（bid）
if a["bid"] > b["ask"] and b["ask"] > 0:
spread_pct = (a["bid"] - b["ask"]) / b["ask"] * 100
if spread_pct >= min_spread_pct:
risk = risk_level(spread_pct * 365, symbol)
rows.append([
f"{symbol}USDT",
f"{EXCHANGES[ex_b]['name']} .4f",
f"{EXCHANGES[ex_a]['name']} .4f",
f".4f",
f".4f",
f"{spread_pct:.4f}%",
risk,
])
rows.sort(key=lambda r: float(r[5].rstrip('%')), reverse=True)
return rows, headers
def main():
import argparse
parser = argparse.ArgumentParser(description="Cross-Exchange Spot Spread Scanner")
parser.add_argument("--format", choices=["table", "markdown", "json"], default="table")
args = parser.parse_args()
rows, headers = scan_spot_spread()
print(f"\n{'='*80}")
print(" Cross-Exchange Spot Spread")
print(f"{'='*80}")
print(format_output(rows, headers, args.format))
print(f"\nFound {len(rows)} opportunities")
if __name__ == "__main__":
main()
FILE:scripts/stablecoin_depeg.py
"""稳定币脱锚监控 — 监测 USDC/DAI 等偏离 $1 的情况"""
from fetcher import _get
from formatter import format_output
# 稳定币监控对，path 用点号分隔访问嵌套 JSON
DEPEG_PAIRS = {
"USDC": [
{"exchange": "Binance", "url": "https://api.binance.com/api/v3/ticker/price", "params": {"symbol": "USDCUSDT"}, "path": "price"},
{"exchange": "OKX", "url": "https://www.okx.com/api/v5/market/ticker", "params": {"instId": "USDC-USDT"}, "path": "data.0.last"},
{"exchange": "Bybit", "url": "https://api.bybit.com/v5/market/tickers", "params": {"category": "spot", "symbol": "USDCUSDT"}, "path": "result.list.0.lastPrice"},
],
"DAI": [
{"exchange": "Binance", "url": "https://api.binance.com/api/v3/ticker/price", "params": {"symbol": "DAIUSDT"}, "path": "price"},
],
"FDUSD": [
{"exchange": "Binance", "url": "https://api.binance.com/api/v3/ticker/price", "params": {"symbol": "FDUSDUSDT"}, "path": "price"},
],
"TUSD": [
{"exchange": "Binance", "url": "https://api.binance.com/api/v3/ticker/price", "params": {"symbol": "TUSDUSDT"}, "path": "price"},
],
}
def _extract_price(data, path: str) -> float:
"""从嵌套 JSON 中按点号路径提取价格"""
keys = path.split(".")
val = data
for k in keys:
if val is None:
print(f" [WARN] 路径解析失败: {path}，在 key={k} 处为 None")
return 0.0
if isinstance(val, list):
val = val[int(k)]
elif isinstance(val, dict):
val = val.get(k)
else:
print(f" [WARN] 路径解析失败: {path}，在 key={k} 处类型异常: {type(val)}")
return 0.0
return float(val) if val else 0.0
def scan_stablecoin_depeg(threshold_pct: float = 0.1) -> tuple:
"""
扫描稳定币脱锚情况
threshold_pct: 偏离阈值百分比（默认 0.1%）
返回 (rows, headers)
"""
headers = ["Stablecoin", "Exchange", "Price (USDT)", "Depeg%", "Status"]
rows = []
print(f"[Stablecoin Depeg] 扫描稳定币脱锚...")
for coin, sources in DEPEG_PAIRS.items():
for src in sources:
data = _get(src["url"], src["params"])
if data is None:
continue
price = _extract_price(data, src["path"])
if price <= 0:
continue
depeg_pct = (price - 1.0) * 100
if abs(depeg_pct) >= threshold_pct:
status = "DEPEGGED" if abs(depeg_pct) > 0.5 else "WATCH"
else:
status = "STABLE"
rows.append([
coin,
src["exchange"],
f".6f",
f"{depeg_pct:+.4f}%",
status,
])
rows.sort(key=lambda r: abs(float(r[3].rstrip('%'))), reverse=True)
return rows, headers
def main():
import argparse
parser = argparse.ArgumentParser(description="Stablecoin Depeg Monitor")
parser.add_argument("--format", choices=["table", "markdown", "json"], default="table")
args = parser.parse_args()
rows, headers = scan_stablecoin_depeg()
print(f"\n{'='*80}")
print(" Stablecoin Depeg Monitor")
print(f"{'='*80}")
print(format_output(rows, headers, args.format))
print(f"\nMonitoring {len(rows)} stablecoin pairs")
if __name__ == "__main__":
main()
FILE:scripts/volume_anomaly.py
"""成交量突变检测 — 发现量能异常放大的币"""
from config import TOP_SYMBOLS
from fetcher import fetch_all_24h_tickers
from formatter import format_output
def scan_volume_anomaly(symbols: list = None, top_n: int = 20) -> tuple:
"""扫描跨所成交量分布，找出某交易所量能异常集中的币种"""
if symbols is None:
symbols = TOP_SYMBOLS
headers = ["Symbol", "Total Vol (USDT)", "Top Exchange", "Vol Share%", "24h Change%", "Signal"]
rows = []
print(f"[Volume Anomaly] 扫描 {len(symbols)} 个交易对...")
for symbol in symbols:
tickers = fetch_all_24h_tickers(symbol)
if len(tickers) < 2:
continue
# 汇总各所成交量
volumes = {}
change_pcts = {}
for exchange, data in tickers.items():
vol = data.get("volume_usdt", 0)
if vol > 0:
volumes[exchange] = vol
change_pcts[exchange] = data.get("change_pct", 0)
if len(volumes) < 2:
continue
total_vol = sum(volumes.values())
if total_vol <= 0:
continue
top_ex = max(volumes, key=volumes.get)
top_share = volumes[top_ex] / total_vol * 100
avg_change = sum(change_pcts.values()) / len(change_pcts)
# 判断信号
if top_share > 70:
signal = "VOLUME SPIKE"
elif top_share > 50 and abs(avg_change) < 2:
signal = "ACCUMULATION"
elif abs(avg_change) > 10:
signal = "MOMENTUM"
else:
signal = "NORMAL"
if signal == "NORMAL":
continue
from config import EXCHANGES
rows.append([
f"{symbol}USDT",
f",.0f",
EXCHANGES[top_ex]["name"],
f"{top_share:.1f}%",
f"{avg_change:+.2f}%",
signal,
])
rows.sort(key=lambda r: float(r[3].rstrip('%')), reverse=True)
return rows[:top_n], headers
def main():
import argparse
parser = argparse.ArgumentParser(description="Volume Anomaly Detection")
parser.add_argument("--format", choices=["table", "markdown", "json"], default="table")
parser.add_argument("--top", type=int, default=20, help="显示前 N 个")
args = parser.parse_args()
rows, headers = scan_volume_anomaly(top_n=args.top)
print(f"\n{'='*80}")
print(" Volume Anomaly Detection")
print(f"{'='*80}")
print(format_output(rows, headers, args.format))
print(f"\nFound {len(rows)} anomalies")
if __name__ == "__main__":
main()