diff --git a/datasource/alltick.py b/datasource/alltick.py new file mode 100644 index 0000000..39639fb --- /dev/null +++ b/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() \ No newline at end of file diff --git a/dingding.py b/dingding.py new file mode 100644 index 0000000..1763329 --- /dev/null +++ b/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) \ No newline at end of file diff --git a/main.py b/main.py index 1b68b49..946fbc0 100644 --- a/main.py +++ b/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: diff --git a/monitors/move.py b/monitors/move.py new file mode 100644 index 0000000..9907213 --- /dev/null +++ b/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) \ No newline at end of file diff --git a/monitors/twitter.py b/monitors/twitter.py new file mode 100644 index 0000000..3a299d8 --- /dev/null +++ b/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) diff --git a/monitors/vegas.py b/monitors/vegas.py index 2efa5ec..2acb36b 100644 --- a/monitors/vegas.py +++ b/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) \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 667e665..7f1619b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,4 +4,5 @@ numpy pandas requests schedule -TA_Lib \ No newline at end of file +TA_Lib +tweepy \ No newline at end of file diff --git a/test.py b/test.py index e954158..5e0fabc 100644 --- a/test.py +++ b/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') \ No newline at end of file +move.run_crypto('1h') \ No newline at end of file