update

datasource/alltick.py

@@ -0,0 +1,31 @@
+import requests
+import urllib.parse
+import json
+
+token = '6c7ba077eee07f6f270e219d4848700e-c-app'
+base_stock_url = 'https://quote.tradeswitcher.com/quote-stock-b-api/batch-kline'
+base_others_url = 'https://quote.tradeswitcher.com/quote-b-api/batch-kline'
+test_headers = {
+    'Content-Type': 'application/json'
+}
+
+def get_klines(type, symbols, interval):
+    query = {
+        "trace": "3baaa938-f92c-4a74-a228-fd49d5e2f8bc-1678419657806",
+        "data": {
+            "code": "AAPL.US",
+            "kline_type": 8,
+            "kline_timestamp_end": 0,
+            "query_kline_num": 500,
+            "adjust_type": 0
+        }
+    }
+
+
+    url = base_stock_url + f"/kline?token={token}"
+    print(url)
+    data = requests.get(url, headers=test_headers).json()
+    print(data)
+
+
+get_klines()

dingding.py

@@ -0,0 +1,9 @@
+import requests
+import setting
+
+url = 'https://oapi.dingtalk.com/robot/send?access_token=c197bfacc77010e92804ecdbb6a6e0f95f86d9e7e4968fb35ec68720eeec8fa8'
+
+# 发送消息
+def send_message(text):
+    formData = {"msgtype":"text","text":{"content": text},"isAtAll":1}
+    requests.post(url, data= formData)

main.py

@@ -13,7 +13,7 @@ times = ["00:00", "04:00", "08:00", "12:00", "16:00", "20:00"]
 for t in times:
     schedule.every().day.at(t).do(vegas.run_crypto, interval = '4h')
 
-version = 'V1.3'
+version = 'V1.4'
 print(f'Running... {version}')
 telegram_sender.send_message(setting.chat_id, f'☕️ Strategy is updated！\r\n\r\nVersion: {version}')
 while True:

monitors/move.py

@@ -0,0 +1,59 @@
+from binance.spot import Spot
+import talib
+import numpy as np
+import pandas as pd
+import datasource.crypto
+import time,setting
+import telegram_sender
+from datasource import crypto
+from datetime import datetime
+import dingding
+
+
+# crossover 函数：检测上穿信号
+def crossover(series1, series2):
+    return (series1 > series2) & (series1.shift(1) <= series2.shift(1))
+
+# crossunder 函数：检测下穿信号
+def crossunder(series1, series2):
+    return (series1 < series2) & (series1.shift(1) >= series2.shift(1))
+
+## 检查信号
+def stratergy_run(symbol, interval, df, debug):
+    maxDifference = 0.005
+
+    ## 计算 ema
+    df['ema5'] = talib.EMA(df['close'], timeperiod=5)
+    df['ema10'] = talib.EMA(df['close'], timeperiod=10)
+    df['ema30'] = talib.EMA(df['close'], timeperiod=30)
+    df['ema144'] = talib.EMA(df['close'], timeperiod=144)
+
+    shortDiff = abs(df['ema5'] - df['ema10']) / df['ema10']
+    midDiff = abs(df['ema10'] - df['ema30']) / df['ema30']
+    longDiff = abs(df['ema30'] - df['ema144']) / df['ema144']
+
+    df['priceCheck'] = (longDiff <= maxDifference) & (midDiff <= maxDifference) & (shortDiff <= maxDifference)
+    df['isLongArrangement'] = (df['ema5'] > df['ema10']) & (df['ema10'] > df['ema30']) & (df['ema30']  > df['ema144'])
+    df['preLongArrangement'] = (df['ema5'].shift(1) > df['ema10'].shift(1)) & (df['ema10'].shift(1) > df['ema30'].shift(1)) & (df['ema30'].shift(1) > df['ema144'].shift(1))
+
+    latest = df.iloc[-1]
+    isbullish = latest['preLongArrangement'] == False & latest['isLongArrangement'] == True & latest['priceCheck'] ==True
+    print(f"{symbol} bullish: {isbullish}")
+    if(isbullish):
+        message = f"⭐️信号策略⭐️\r\n\r\n"
+        message += f"策略:【趋势追踪V1.0】\r\n"
+        message += f"品种: {symbol}\r\n"
+        message += f"周期: {interval}\r\n\r\n"
+        message += f"信号: 【启动】\r\n\r\n"   
+
+        dingding.send_message(message)
+
+
+def run_crypto(interval, debug=False):
+
+    symbols = ['BTCUSDT',"ETHUSDT",'LTCUSDT','DOGEUSDT','FTMUSDT','FILUSDT','OPUSDT','SOLUSDT','BNBUSDT','BCHUSDT','ETCUSDT','ARUSDT']
+    for s in symbols:
+        df = crypto.get_klines(s, interval)
+        stratergy_run(s,interval, df, debug)
+        
+        time.sleep(1)

monitors/twitter.py

@@ -0,0 +1,42 @@
+import tweepy
+import time
+
+# 替换为你的 Twitter API 密钥
+API_KEY = 'PvmpsQ03urOzrZLxFRaTlq1e2'
+API_KEY_SECRET = 'SY2xcGOJBVRO8pStF23Al5lHgXRJ95b9gzHuOdX6sI40D9IZ0Z'
+ACCESS_TOKEN = '209453217-1TBd7HtWSrzVc24ZYVzb3JwbOtxPxGuqRwq4bXuW'
+ACCESS_TOKEN_SECRET = 'oU39wnIOCwILHZ2RXbzHSGnS9e0SAsvtwF6PhyhDU0Hra'
+BEARER_TOKEN = 'AAAAAAAAAAAAAAAAAAAAAN3VtQEAAAAAKSMZ3VcPmGVwMgxIjX8yeb9Fbow%3DXl2BW9tUi5vn1y6uFpcA8DsyXp56yd0MQnmGhheWK04tFKPxDf'
+
+# 认证到 Twitter API
+client = tweepy.Client(bearer_token=BEARER_TOKEN, 
+                       consumer_key=API_KEY, 
+                       consumer_secret=API_KEY_SECRET,
+                       access_token=ACCESS_TOKEN, 
+                       access_token_secret=ACCESS_TOKEN_SECRET)
+
+# 指定要监控的 Twitter 账号
+twitter_username = 'whale_alert'
+
+# 获取用户ID
+user = client.get_user(username=twitter_username)
+user_id = user.data.id
+
+# 已处理推文的ID集合
+processed_tweet_ids = set()
+
+def monitor_tweets(user_id):
+    # 获取该用户最新的推文
+    response = client.get_users_tweets(id=user_id, max_results=5)
+    tweets = response.data
+
+    # 处理和打印新的推文
+    for tweet in tweets:
+        if tweet.id not in processed_tweet_ids:
+            print(f'{tweet.id} - {tweet.text}')
+            processed_tweet_ids.add(tweet.id)
+
+# 定期监控推文（例如每隔 60 秒）
+while True:
+    monitor_tweets(user_id)
+    time.sleep(60)

monitors/vegas.py

@@ -7,6 +7,7 @@ import time,setting
 import telegram_sender
 from datasource import crypto
 from datetime import datetime
+import dingding
 
 
 # crossover 函数：检测上穿信号
@@ -18,7 +19,7 @@ def crossunder(series1, series2):
     return (series1 < series2) & (series1.shift(1) >= series2.shift(1))
 
 ## 检查信号
-def stratergy_run(symbol, interval,df):
+def stratergy_run(symbol, interval, df, debug):
     ## 计算 ema
     df['ema13'] = talib.EMA(df['close'], timeperiod=13)
     df['ema144'] = talib.EMA(df['close'], timeperiod=144)
@@ -63,16 +64,22 @@ def stratergy_run(symbol, interval,df):
 
     print(f"【{symbol} - {interval}】 is checked!")
     if direction != "":
-        telegram_sender.send_message(setting.chat_id, message)
-        print(f"【{symbol} - {interval}】 is singaled!")
+        if debug == False:
+            telegram_sender.send_message(setting.chat_id, message)
+            dingding.send_message(message)
+            print(f"【{symbol} - {interval}】 is singaled!")
+        else:
+            print(message)
 
-def run_crypto(interval):
+
+def run_crypto(interval, debug=False):
     print('Vegas策略运行.')
 
+    # symbols = crypto.get_symbols()
     # symbols = crypto.get_top_binance_usdt_pairs(10)
     symbols = ['BTCUSDT',"ETHUSDT",'LTCUSDT','DOGEUSDT','FTMUSDT','FILUSDT','OPUSDT','SOLUSDT','BNBUSDT','BCHUSDT','ETCUSDT','ARUSDT']
     for s in symbols:
         df = crypto.get_klines(s, interval)
-        stratergy_run(s,interval, df)
+        stratergy_run(s,interval, df, debug)
         
         time.sleep(1)

requirements.txt

@@ -4,4 +4,5 @@ numpy
 pandas
 requests
 schedule
-TA_Lib
+TA_Lib
+tweepy

test.py

@@ -1,8 +1,4 @@
-import yfinance as yf
-import pandas as pd
-from datasource import crypto
+import requests
+from monitors import move
 
-from monitors import large_transfer
-from monitors import vegas
-
-vegas.run_crypto('4h')
+move.run_crypto('1h')