-
Notifications
You must be signed in to change notification settings - Fork 0
/
telegram.py
164 lines (123 loc) · 4.75 KB
/
telegram.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
from telethon.tl.types import PeerChannel
from telethon import TelegramClient
from dotenv import load_dotenv
from os import getenv
import re
from datetime import timedelta, datetime
from time import localtime
from debug import log
from json import load
load_dotenv()
api_id = getenv('TELEGRAM_API_ID')
api_hash = getenv('TELEGRAM_API_HASH')
phone = getenv('TELEGRAM_PHONE')
password = getenv('TELEGRAM_PASSWORD')
client = TelegramClient('tradingbot', api_id, api_hash).start(
phone=phone, password=password)
# copy telegram_channels.example.json to provide the channels
CHANNELS = [{**channel, 'id': PeerChannel(channel_id=channel['id'])}
for channel in load(open('telegram_channels.json', 'r'))]
def get_message_options(message):
"""
Parses a telegram message and retrieves all information necessary for scheduling a trade
Obs: this implementation is specific for my scenario. Other message patterns will require changes in this funcion
"""
action_map = {'COMPRA': 'call', 'VENDA': 'put'}
start_time = None
pair = None
action = None
timeframe = None
log(f'Message received: {message}', False)
try:
start_time = re.search('(\d{2}):(\d{2})', message)
if start_time:
start_time = f'{start_time.group(1)}:{start_time.group(2)}'
pair = re.search('Moeda: (\w{6}|\w{3}/\w{3})\n', message)
if not pair:
pair = re.search('(\w{6});', message)
if not pair:
pair = re.search('(\w{3}/\w{3})\s?', message)
if not pair:
pair = re.search('\s(\w{6})-', message)
if not pair:
pair = re.search('\s(\w{6})\(', message)
if not pair:
pair = re.search('\s(\w{6})\s', message)
pair = pair.group(1).replace('/', '')
otc = re.search('(OTC)', message)
otc = otc or re.search('-OTC', message)
if otc:
pair += '-OTC'
action = re.search(';(COMPRA|VENDA)', message)
if not action:
action = re.search('Sinal - .(COMPRA|VENDA).\n', message)
else:
action = action_map[action.group(1)]
if not action:
action = re.search('PUT|CALL', message)
action = action.group(0).lower()
timeframe = re.search('Timeframe M(\d)', message)
if not timeframe:
timeframe = re.search('Expiração(?: M|: )(\d)', message)
if timeframe:
timeframe = int(timeframe.group(1))
else:
timeframe = 5
except Exception as e:
log(f'Error while parsing message: {e}', False)
return None
log(
f'Parsed pair, action, start time and timeframe: {(pair, action, start_time, timeframe)}', False)
return (pair, action, start_time, timeframe)
def get_message_options_list(message):
"""
Parses a telegram message and retrieves all information necessary for scheduling multiple trades
Obs: this implementation is specific for my scenario. Other message patterns will require changes in this funcion
"""
options_list = []
for line in message.split('\n'):
start_time = None
duration = None
pair = None
action = None
try:
start_time = re.search('(\d{2}):(\d{2})', line)
duration = timedelta(hours=int(start_time.group(1)), minutes=int(start_time.group(
2))) - timedelta(hours=localtime().tm_hour, minutes=localtime().tm_min, seconds=localtime().tm_sec)
pair = re.search('(\w{6}|\w{3}/\w{3})', line)
pair = pair.group(1).replace('/', '')
action = re.search('(PUT|CALL)', line)
action = action.group(1).lower()
except Exception as e:
continue
if duration and duration.seconds / 3600 > 9:
continue
log(
f'Parsed pair, action and start time: {(pair, action, duration.seconds)}', False)
options_list.append((pair, action, duration.seconds))
return options_list if len(options_list) > 0 else None
def run_command(command):
commands = {
'logs': logs,
'stop': stop
}
args = command.split(' ')
command = args[0]
args = args[1:]
return commands[command](*args) if command in commands else None
def logs(page=None):
limit = 4096
page = int(page) if page else 1
with open(f'./log-{datetime.today().strftime("%d-%m-%Y")}.txt', 'r', encoding='utf-8') as f:
content = f.read()
if len(content) > limit:
content = content[0 + limit * (page - 1):limit * page]
if len(content) == limit:
return content[:-3] + '...'
else:
return content
else:
return content
def stop():
print('Stop command received')
return 'STOP'