ianzeng123
암호화폐 추세 지표 개발 기록에 대한 간략한 논의
0
516
기술적 분석 분야에서 “고가(HH)”, “저가(HL)”, “저가(LH)”, “저가(LL)“의 네 가지 핵심 가격 구조 패턴을 파악하는 것은 시장 추세 방향과 잠재적 반전 시점을 판단하는 데 있어 핵심적인 요소입니다. 이러한 패턴은 시장 수요와 공급의 역동적인 균형과 지배적인 심리(강세 또는 약세)를 직관적으로 보여주어 거래 결정에 객관적인 근거를 제공합니다.
강세장과 약세장의 핵심 특징
강세 추세: 고점과 저점이 높아지는 것은 강세 추세의 핵심 지표입니다. 고점이 높아지는 것은 가격 고점이 이전 고점을 넘어설 때 발생하며, 이는 매수자들이 가격을 끌어올리고 있음을 나타내며 시장 강세를 반영합니다. 저점이 높아지는 것은 가격 하락이 이전 하락보다 높은 수준에서 멈출 때 발생하며, 이는 시장이 상승 모멘텀을 유지하고 있음을 나타냅니다. 이러한 패턴들은 차트에서 일련의 상승 고점과 저점을 통해 확인되는 강력한 상승 추세를 나타냅니다.
약세 추세: 고점과 저점이 낮아지면 하락 추세를 나타냅니다. 고점이 낮아지면 가격 고점이 이전 고점 수준에 도달하지 못하여 매수 압력이 약화됨을 나타냅니다. 저점이 낮아지면 가격이 이전 저점 아래로 하락하여 매도 압력이 증가하고 시장이 약세를 보임을 나타냅니다. 이러한 패턴은 가격 차트에서 일련의 하락 고점과 저점을 통해 나타나는 하락 추세를 파악하는 데 필수적입니다.
정량적 추세 식별의 필요성
암호화폐 시장은 높은 변동성, 24시간 연중무휴 거래, 그리고 시장 심리에 따른 중요한 추세를 특징으로 합니다. 이러한 환경에서는 추세 패턴을 정확하게 파악하는 것이 더욱 중요해집니다. “고가, 저가” 또는 “고가, 저가”의 연속성을 정량화함으로써 시장 추세를 더욱 정확하게 파악하고 거래 결정에 객관적인 근거를 제공할 수 있습니다.
FMZ 플랫폼을 선택해야 하는 이유
FMZ Inventor 양적 플랫폼은 이러한 지표 개발을 위한 이상적인 환경을 제공합니다.
데이터 이점：
- 주요 거래소의 과거 데이터 무료 제공
- 주요 암호화폐를 포괄하는 완전한 K-라인 데이터
- 신뢰할 수 있는 데이터 품질과 시기적절한 업데이트
개발 환경：
- 가볍고 빠른 코드 편집 페이지
- Python 등 다양한 프로그래밍 언어 지원
- 풍부한 기술 분석 기능 라이브러리 내장
편의성 테스트：
- 완전한 백테스팅 기능
- 실시간 모니터링 및 시각화
- 여러 통화를 동시에 분석하는 데 편리합니다.
이러한 장점을 바탕으로 FMZ 플랫폼은 고가와 저가 연속성 추세 지표를 탐색하는 데 선택되었습니다.
암호화폐 시장의 특징
지표를 설계하기 전에 암호화폐 시장과 주식 시장의 차이점을 고려해야 합니다.
- 24시간 거래, 시장 폐쇄 없음
- 변동성이 엄청나서, 시장이 하루에 20%나 오르거나 내리는 일은 드문 일이 아닙니다.
- 개인투자자가 많고 감정적 거래가 당연하다
- 역사적 데이터는 길지 않으며 대부분의 통화는 몇 년밖에 없습니다.
이러한 특성을 바탕으로 설계안은 다음과 같다.
- 일일 데이터를 사용하면 뒤처지지 않고 일중 노이즈를 걸러낼 수 있습니다.
- 정확성과 적시성의 균형을 맞추기 위해 최소 확인 기간을 3일로 설정하세요.
- 동시에 여러 주요 통화를 모니터링하여 검증합니다.
FMZ 플랫폼에 구현됨
핵심 알고리즘 설계
심도 있는 고찰 끝에, 당일 불완전한 데이터로 인한 잘못된 판단을 방지하기 위해 “어제의 완전한 데이터”를 기반으로 한 분석 방법을 채택했습니다. 핵심 데이터 구조는 다음과 같습니다.
# 每个币种的数据都单独存储
data = defaultdict(lambda: {
"daily_records": [], # 存储每日的昨天数据
"trend_buffer": [], # 当前趋势缓冲区
"patterns": [], # 完整的趋势模式
"current_trend": None, # 当前趋势状态
"last_processed_time": 0
})
트렌드 식별의 핵심 논리
추세를 파악하기 위한 주요 기능:
def is_trend_continuing(self, buffer, trend_type):
"""检查趋势是否持续"""
if len(buffer) < 2:
return False
curr = buffer[-1]
prev = buffer[-2]
if trend_type == "BULL":
# 牛市：High和Low都上升
return curr["High"] > prev["High"] and curr["Low"] > prev["Low"]
elif trend_type == "BEAR":
# 熊市：High和Low都下降
return curr["High"] < prev["High"] and curr["Low"] < prev["Low"]
return False
추세 상태 관리:
def analyze_trend_state(self, symbol):
"""分析趋势状态"""
storage = data[symbol]
buffer = storage["trend_buffer"]
current_trend = storage["current_trend"]
if current_trend is None:
# 尝试检测新趋势
new_trend = self.detect_new_trend(buffer)
if new_trend:
storage["current_trend"] = {
"type": new_trend,
"start_time": buffer[-2]["Time"],
"start_price": buffer[-2]["Close"],
"consecutive_days": 1
}
else:
# 检查现有趋势是否继续
if self.is_trend_continuing(buffer, current_trend["type"]):
current_trend["consecutive_days"] += 1
else:
# 趋势中断，记录完整模式
if current_trend["consecutive_days"] >= MIN_CONSECUTIVE:
# 保存趋势记录
self.save_pattern(symbol, current_trend, buffer)
핵심적인 디자인 아이디어는 높은 점과 낮은 점을 요구하는 것입니다.동시에지속적인 변동을 충족하고 최소 3일의 확인 기간을 확보하면 오판을 크게 줄일 수 있습니다. 수익률 통계는 추세 시작 시점의 시가에서 추세 종료 시점의 종가까지의 증가 또는 감소를 나타냅니다.
실제 운영 결과 및 데이터 분석
2020년부터 2025년 6월까지 FMZ 플랫폼의 과거 데이터 백테스트를 기반으로, 지난 10개 완전한 추세 주기에서 3대 주요 통화의 실제 성과는 다음과 같습니다.
ETH 테스트 결과 분석
| 유형 | 시작 날짜 | 종료일 | 지속 | 생산하다 |
|---|---|---|---|---|
| 맥주 시장 | 2025-05-29 | 2025-06-01 | 3 | -5.38% |
| 강세장 | 2025-05-19 | 2025-05-22 | 3 | 6.73% |
| 강세장 | 2025-05-06 | 2025-05-09 | 3 | 26.94% |
| 강세장 | 2025-04-24 | 2025-04-27 | 3 | -0.17% |
| 맥주 시장 | 2025-03-25 | 2025-03-30 | 5 | -13.13% |
| 강세장 | 2025-03-21 | 2025-03-24 | 3 | 5.04% |
| 맥주 시장 | 2025-01-06 | 2025-01-10 | 4 | -10.86% |
| 강세장 | 2025-01-01 | 2025-01-06 | 5 | 11.2% |
| 맥주 시장 | 2024-12-17 | 2024-12-20 | 3 | -15.5% |
| 맥주 시장 | 2024-12-07 | 2024-12-10 | 3 | -9.96% |
ETH 성능 특성：
- 가장 두드러진 성과는 5월 6일부터 5월 9일까지의 강세장으로, 단 3일 만에 26.94%나 급등했습니다.
- 평균적인 강세장은 3.4일간 지속되며, 평균 수익률은 9.97%입니다.
- 평균 하락장은 3.6일간 지속되며, 평균 하락률은 -10.97%입니다.
- 극도로 변동성이 크고 3대 통화 중 가장 불안정합니다.
BTC 테스트 결과 분석
| 유형 | 시작 날짜 | 종료일 | 지속 | 생산하다 |
|---|---|---|---|---|
| 강세장 | 2025-06-06 | 2025-06-11 | 5 | 7.78% |
| 맥주 시장 | 2025-06-03 | 2025-06-06 | 3 | -0.78% |
| 맥주 시장 | 2025-05-27 | 2025-05-31 | 4 | -4.37% |
| 맥주 시장 | 2025-05-22 | 2025-05-25 | 3 | -2.63% |
| 강세장 | 2025-05-06 | 2025-05-09 | 3 | 8.4% |
| 맥주 시장 | 2025-05-02 | 2025-05-05 | 3 | -2.37% |
| 강세장 | 2025-04-20 | 2025-04-23 | 3 | 10.07% |
| 강세장 | 2025-04-09 | 2025-04-13 | 4 | 10.25% |
| 맥주 시장 | 2025-03-26 | 2025-03-29 | 3 | -5.53% |
| 맥주 시장 | 2025-03-08 | 2025-03-11 | 3 | -5.81% |
BTC 성능 특성：
- 10주기 중 6주기가 약세장인 약세장이 지배적입니다.
- 평균적인 강세장은 3.75일 동안 지속되며, 평균 수익률은 9.13%입니다.
- 평균 하락장은 3.17일 동안 지속되었고 평균 하락률은 -3.58%였습니다.
- 전반적인 성과는 비교적 균형 잡혀 있으며 변동성은 적당합니다.
BNB 테스트 결과 분석
| 유형 | 시작 날짜 | 종료일 | 지속 | 생산하다 |
|---|---|---|---|---|
| 강세장 | 2025-06-06 | 2025-06-11 | 5 | 5.46% |
| 맥주 시장 | 2025-06-03 | 2025-06-06 | 3 | -2.73% |
| 강세장 | 2025-05-19 | 2025-05-22 | 3 | 4.63% |
| 강세장 | 2025-05-05 | 2025-05-10 | 5 | 11.95% |
| 강세장 | 2025-04-20 | 2025-04-23 | 3 | 2.44% |
| 강세장 | 2025-04-09 | 2025-04-12 | 3 | 7.63% |
| 강세장 | 2025-03-14 | 2025-03-17 | 3 | 8.18% |
| 맥주 시장 | 2025-03-08 | 2025-03-11 | 3 | -7.49% |
| 강세장 | 2025-02-10 | 2025-02-13 | 3 | 9.66% |
| 맥주 시장 | 2025-01-31 | 2025-02-03 | 3 | -12.2% |
BNB 성능 특성：
- 10주기 중 7주기가 강세장인 강세장이 지배적입니다.
- 평균적인 강세장은 3.43일간 지속되었고, 평균 수익률은 7.14%였습니다.
- 평균 하락장은 3일간 지속되며 평균 하락률은 -7.47%입니다.
- 성과는 가장 안정적이며, 극단적인 시장 상황이 상대적으로 적습니다.
데이터 뒤에 숨겨진 흥미로운 발견
이 3개 통화의 최근 10개 추세 주기 데이터를 분석한 결과, 몇 가지 흥미로운 현상이 발견되었습니다.
추세의 지속 기간에 관하여
대부분의 추세는 약 3~5일 안에 끝나는데, 이는 암호화폐 시장에 대한 모든 사람들의 생각과 일맥상통합니다. 시장은 매우 빠르게 변하기 때문입니다. 최소 확인 기간을 3일로 설정한 초기 설정은 현재에도 상당히 합리적인데, 하루 중 발생하는 무작위적인 변동을 걸러내고 너무 오래 기다리다 기회를 놓치는 일이 없기 때문입니다. 비트코인은 이러한 측면에서 가장 안정적이며, 추세 지속 기간도 비교적 규칙적입니다.
다양한 통화의 “성격”의 차이점
이 세 가지 암호화폐는 각기 다른 특징을 가지고 있습니다. ETH의 최근 상승세는 더욱 눈길을 끌며, 너무 오랫동안 쪼그리고 앉아 있었기 때문에 반등 변동성이 매우 큽니다. 5월 6일부터 9일까지 3일 만에 26.94%까지 상승할 수 있다는 것은 놀라운 일이지만, -0.17%의 “강세장”도 있어 사람들이 어리둥절해하는 상황입니다. BTC는 의심할 여지 없이 더 안정적입니다. 최근 약세장이 더 많이 발생했지만, 변동성은 여전히 용납할 수 있는 수준입니다. BNB는 많은 놀라움을 선사했는데, 강세장이 70%를 차지하며 위험 대비 수익률(R/R)이 가장 좋은 것으로 보입니다.
추세 판단에 대한 몇 가지 관찰
결과를 보면, 이 간단한 지표는 몇 가지 중요한 순간을 포착하고 있습니다. 예를 들어 ETH의 26.94% 급등, BTC와 BNB의 여러 차례의 강세장 주기, 그리고 약세장을 예고하는 몇 가지 시의적절한 신호들이 있습니다. 물론 -0.17%의 “강세장”과 같이 알고리즘이 아직 개선의 여지가 있음을 보여주는 몇 가지 혼란스러운 지점도 있습니다.
이 도구는 무슨 용도인가요?
그것이 당신을 위해 할 수 있는 일
솔직히 말해서 이 도구는 주로 당신에게 도움이 됩니다.지금 시장이 어떤지 알아보세요：
- 현재 추세가 상승하고 있는지, 하락하고 있는지, 아니면 옆으로 변동하고 있는지 알려줍니다.
- 이러한 추세가 얼마나 오랫동안 지속되었는지, 그리고 어떤 성과를 보이고 있는지 추적하세요.
- 전적으로 감정에 근거하지 않고 상대적으로 객관적인 판단 기준을 제공합니다.
- 실제 데이터를 보면 3~5일간의 단기 추세를 파악하는 데 매우 효과적입니다.
그것이 할 수 없는 것
이 도구는 다음과 같은 점을 명확히 해야 합니다.미래를 예측하기 위한 것은 아닙니다.：
- 내일 가격이 오를지 내릴지는 알려주지 않습니다.
- 얼마나 오르거나 내릴지 예측할 수 없습니다.
- 이는 귀하의 위험 관리 및 자금 관리를 대체할 수 없습니다.
- 시장이 옆으로 움직일 때 혼란스러운 신호가 나타날 수 있습니다.
사용 중 발생한 몇 가지 문제
실제 운영에서는 다음과 같은 몇 가지 제한 사항도 발견되었습니다.
응답이 좀 느립니다: 확인까지 3일이 걸리기 때문에 처음 며칠 동안은 추세를 파악하기 사실상 불가능하다.
가끔 나는 “잘못된 사람을 본다”：ETH의 -0.17% “상승장”과 마찬가지로, 일부 특수한 경우 알고리즘의 판단이 잘못될 수 있음을 보여줍니다.
횡보장은 골칫거리다：시장이 일정 범위 내에서 변동할 경우 신호가 자주 바뀌어 불편할 수 있습니다.
가격만 보면 좀 단조롭네요: 거래량, 뉴스 등 동등하게 중요한 요소를 고려하지 못함
다음에는 어떻게 개선할 수 있을까요?
이 기간 동안의 관찰을 바탕으로, 시도해 볼 만한 방향이 몇 가지 있다고 생각합니다.
다양한 통화에 대한 매개변수 조정： ETH와 같이 변동성이 큰 통화의 경우 더 엄격한 확인 조건이 필요할 수 있지만, BNB와 같이 비교적 안정적인 통화의 경우 확인 시간을 단축할 수 있습니다. 또한, 수익률이 너무 낮은 신호를 걸러내기 위해 최소 수익률 임계값을 설정할 수도 있습니다.
보조 판단을 추가하세요： 예를 들어, 추세가 신뢰할 만한지 확인하기 위해 거래량의 변화를 결합하거나, 사소한 변화에 의해 오도되는 것을 피하기 위해 가격 변동 범위를 고려합니다.
알고리즘 자체 최적화： 추세 중단의 판단 논리를 개선하여 잘못된 판단을 줄입니다. 추세에 강도 평가를 추가하여 강한 추세와 약한 추세를 구분합니다. 일부 비정상 상황에 대한 특별한 처리 메커니즘을 구축합니다.
이 탐험을 되돌아보며
이 간단한 시장 모니터링 도구는 전통적인 기술 분석 개념을 자동화된 시스템으로 전환합니다. FMZ 플랫폼의 편리함을 활용하여 암호화폐 시장의 상황을 실시간으로 모니터링할 수 있는 도구를 성공적으로 구축했습니다.
이 보고서의 주요 가치는 시장 상황에 대한 비교적 객관적인 기록을 제공하는 데 있으며, 이를 통해 다음과 같은 이점을 얻을 수 있습니다.
- 전체 시장 상황에 대한 거시적 이해를 갖습니다.
- 과거 데이터를 통해 일부 인기 통화 필터링
- 더욱 심층적인 분석을 위한 데이터 지원 제공
데이터가 계속 축적됨에 따라 이 도구의 가치는 점점 더 높아질 것입니다. 물론 이는 수많은 분석 도구 중 하나일 뿐이며 모든 문제를 해결할 수 있을 것이라고 기대할 수는 없지만, 시작점으로서는 여전히 매우 흥미롭다고 생각합니다.
'''backtest
start: 2020-01-01 00:00:00
end: 2025-06-16 00:00:00
period: 1d
basePeriod: 1d
exchanges: [{"eid":"Binance","currency":"BTC_USDT"}]
'''
import json
from datetime import datetime
from collections import defaultdict
# 配置参数
SYMBOLS = ["ETH_USDT", "BTC_USDT", "BNB_USDT"]
MIN_CONSECUTIVE = 3 # 最少连续天数
MAX_HISTORY = 1000 # 最大历史记录数
# 全局数据存储
data = defaultdict(lambda: {
"daily_records": [], # 存储每日的昨天数据
"trend_buffer": [], # 当前趋势缓冲区
"patterns": [], # 完整的趋势模式
"current_trend": None, # 当前趋势状态
"last_processed_time": 0
})
class TrendAnalyzer:
def get_yesterday_data(self, records):
"""获取昨天的完整数据（records[-2]）"""
if len(records) < 2:
return None
return records[-2] # 昨天的完整K线数据
def is_trend_continuing(self, buffer, trend_type):
"""检查趋势是否持续"""
if len(buffer) < 2:
return False
curr = buffer[-1]
prev = buffer[-2]
if trend_type == "BULL":
# 牛市：High和Low都上升
return curr["High"] > prev["High"] and curr["Low"] > prev["Low"]
elif trend_type == "BEAR":
# 熊市：High和Low都下降
return curr["High"] < prev["High"] and curr["Low"] < prev["Low"]
return False
def detect_new_trend(self, buffer):
"""从缓冲区检测新趋势"""
if len(buffer) < 2:
return None
curr = buffer[-1]
prev = buffer[-2]
# 检查是否开始牛市趋势
if curr["High"] > prev["High"] and curr["Low"] > prev["Low"]:
return "BULL"
# 检查是否开始熊市趋势
elif curr["High"] < prev["High"] and curr["Low"] < prev["Low"]:
return "BEAR"
return None
def process_daily_data(self, symbol, records):
"""处理每日数据"""
if not records or len(records) < 2:
return
storage = data[symbol]
yesterday_data = self.get_yesterday_data(records)
if not yesterday_data or yesterday_data["Time"] <= storage["last_processed_time"]:
return # 没有新的昨天数据
# 更新处理时间
storage["last_processed_time"] = yesterday_data["Time"]
# 添加到每日记录
storage["daily_records"].append(yesterday_data)
if len(storage["daily_records"]) > MAX_HISTORY:
storage["daily_records"] = storage["daily_records"][-MAX_HISTORY:]
# 添加到趋势缓冲区
storage["trend_buffer"].append(yesterday_data)
# 分析趋势
self.analyze_trend_state(symbol)
def analyze_trend_state(self, symbol):
"""分析趋势状态"""
storage = data[symbol]
buffer = storage["trend_buffer"]
current_trend = storage["current_trend"]
if len(buffer) < 2:
return
if current_trend is None:
# 尝试检测新趋势
new_trend = self.detect_new_trend(buffer)
if new_trend:
storage["current_trend"] = {
"type": new_trend,
"start_time": buffer[-2]["Time"], # 趋势从前一天开始
"start_price": buffer[-2]["Close"],
"start_open": buffer[-2]["Open"],
"consecutive_days": 1
}
Log(f"{symbol} 检测到{new_trend}趋势开始")
else:
# 没有趋势，只保留最近的数据
storage["trend_buffer"] = buffer[-1:]
else:
# 检查现有趋势是否继续
if self.is_trend_continuing(buffer, current_trend["type"]):
# 趋势继续
current_trend["consecutive_days"] += 1
# 检查是否达到最小天数要求
if current_trend["consecutive_days"] == MIN_CONSECUTIVE:
trend_name = "牛市" if current_trend["type"] == "BULL" else "熊市"
Log(f"{symbol} {trend_name}趋势确认! 连续{MIN_CONSECUTIVE}天")
else:
# 趋势中断
if current_trend["consecutive_days"] >= MIN_CONSECUTIVE:
# 记录完整的趋势
end_data = buffer[-2] # 趋势在前一天结束
duration = current_trend["consecutive_days"]
start_price = current_trend["start_open"]
end_price = end_data["Close"]
return_pct = round((end_price - start_price) / start_price * 100, 2)
storage["patterns"].append({
"trend": current_trend["type"],
"start_time": current_trend["start_time"],
"end_time": end_data["Time"],
"duration": duration,
"return": return_pct
})
trend_name = "牛市" if current_trend["type"] == "BULL" else "熊市"
Log(f"{symbol} {trend_name}趋势结束，持续{duration}天，收益{return_pct}%")
# 重置趋势状态，重新开始检测
storage["current_trend"] = None
storage["trend_buffer"] = buffer[-2:] # 保留最近两天数据重新开始
# 立即检测新趋势
self.analyze_trend_state(symbol)
def generate_tables():
"""生成所有统计表格"""
tables = []
# 概览表
overview_rows = []
for symbol in SYMBOLS:
storage = data[symbol]
if not storage["daily_records"]:
continue
patterns = storage["patterns"]
current_trend = storage["current_trend"]
# 计算统计数据
bull_patterns = [p for p in patterns if p["trend"] == "BULL"]
bear_patterns = [p for p in patterns if p["trend"] == "BEAR"]
stats = {
"bull_avg_return": round(sum(p["return"] for p in bull_patterns) / len(bull_patterns), 2) if bull_patterns else 0,
"bear_avg_return": round(sum(p["return"] for p in bear_patterns) / len(bear_patterns), 2) if bear_patterns else 0,
"bull_avg_days": round(sum(p["duration"] for p in bull_patterns) / len(bull_patterns), 1) if bull_patterns else 0,
"bear_avg_days": round(sum(p["duration"] for p in bear_patterns) / len(bear_patterns), 1) if bear_patterns else 0
}
# 当前状态
current_status = "震荡"
current_return = 0
current_days = 0
consecutive = 0
if current_trend and storage["daily_records"]:
latest_price = storage["daily_records"][-1]["Close"]
start_price = current_trend["start_open"]
current_return = round((latest_price - start_price) / start_price * 100, 2)
current_days = current_trend["consecutive_days"]
current_status = "牛市" if current_trend["type"] == "BULL" else "熊市"
consecutive = current_trend["consecutive_days"]
overview_rows.append([
symbol.replace("_USDT", ""),
current_status,
str(current_days),
f"{current_return}%",
str(consecutive),
str(len(bull_patterns)),
str(len(bear_patterns)),
f"{stats['bull_avg_return']}%",
f"{stats['bear_avg_return']}%",
f"{stats['bull_avg_days']}天",
f"{stats['bear_avg_days']}天"
])
tables.append({
"type": "table",
"title": "每日高低价趋势监控（基于昨日完整数据）",
"cols": ["币种", "状态", "持续", "收益", "强度", "牛市次数", "熊市次数", "牛市均收益", "熊市均收益", "牛市均天数", "熊市均天数"],
"rows": overview_rows
})
# 趋势缓冲区分析表
buffer_rows = []
for symbol in SYMBOLS:
storage = data[symbol]
buffer = storage["trend_buffer"]
current_trend = storage["current_trend"]
if not buffer:
continue
latest_price = buffer[-1]["Close"]
buffer_size = len(buffer)
# 显示最近几天的High/Low变化
if len(buffer) >= 2:
recent_highs = [f"{r['High']:.0f}" for r in buffer[-min(5, len(buffer)):]]
recent_lows = [f"{r['Low']:.0f}" for r in buffer[-min(5, len(buffer)):]]
high_trend = " → ".join(recent_highs)
low_trend = " → ".join(recent_lows)
else:
high_trend = f"{buffer[-1]['High']:.0f}"
low_trend = f"{buffer[-1]['Low']:.0f}"
trend_status = "无趋势"
if current_trend:
trend_status = f"{'牛市' if current_trend['type'] == 'BULL' else '熊市'}{current_trend['consecutive_days']}天"
buffer_rows.append([
symbol.replace("_USDT", ""),
f"{latest_price:.2f}",
trend_status,
str(buffer_size),
high_trend,
low_trend
])
tables.append({
"type": "table",
"title": "趋势缓冲区状态",
"cols": ["币种", "价格", "当前趋势", "缓冲区", "High变化", "Low变化"],
"rows": buffer_rows
})
# 历史记录表
for symbol in SYMBOLS:
patterns = [p for p in data[symbol]["patterns"] if p["duration"] >= MIN_CONSECUTIVE]
coin_name = symbol.replace("_USDT", "")
if not patterns:
tables.append({
"type": "table",
"title": f"{coin_name} 历史趋势",
"cols": ["类型", "开始", "结束", "天数", "收益"],
"rows": [["无数据", "-", "-", "-", "-"]]
})
continue
rows = []
for p in sorted(patterns, key=lambda x: x["end_time"], reverse=True)[:10]: # 只显示最近10条
rows.append([
"牛市" if p["trend"] == "BULL" else "熊市",
datetime.fromtimestamp(p["start_time"] / 1000).strftime('%Y-%m-%d'),
datetime.fromtimestamp(p["end_time"] / 1000).strftime('%Y-%m-%d'),
str(p["duration"]),
f"{p['return']}%"
])
tables.append({
"type": "table",
"title": f"{coin_name} 历史趋势",
"cols": ["类型", "开始", "结束", "天数", "收益"],
"rows": rows
})
return tables
def main():
analyzer = TrendAnalyzer()
Log("趋势分析系统启动 - 基于昨日完整数据的逐日分析")
Log("牛市定义: High和Low连续上升≥3天")
Log("熊市定义: High和Low连续下降≥3天")
while True:
try:
# 处理每个币种的数据
for symbol in SYMBOLS:
records = exchange.GetRecords(symbol)
analyzer.process_daily_data(symbol, records)
# 生成并显示表格
tables = generate_tables()
LogStatus('`' + json.dumps(tables) + '`')
except Exception as e:
Log(f"错误: {str(e)}")
Sleep(1000 * 60 * 60) # 24小时
def onexit():
total = sum(len(data[s]["patterns"]) for s in SYMBOLS)
Log(f"系统停止, 共识别 {total} 个趋势模式")