url_ws = wss://data.tradingview.com/socket.io/websocket
def get_auth_token(self):
data = {"username": "your_tradingview_username", "password": "your_tradingview_password", "remember": "on"}
headers = {'Referer': 'https://ru.tradingview.com', 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.198 Safari/537.36','authority': 'ru.tradingview.com','x-language': 'ru'}
response = requests.post(url="https://www.tradingview.com/accounts/signin/", data=data, headers=headers)
auth_token = response.json()['user']['auth_token']
return auth_token
def on_open(self):
self.ws.send(self.generate_message('set_auth_token', [self.get_auth_token()]))
dict_session = self.create_dict_of_symbol()
one_session_dict_list = []
for one_session_dict in dict_session:
cs = one_session_dict[list(one_session_dict.keys())[0]]['cs']
self.ws.send(self.generate_message('chart_create_session', [cs, ""]))
resolve_symbol = "={"symbol":"" + list(one_session_dict.keys())[0] + "","adjustment":"splits"}"
self.ws.send(self.generate_message('resolve_symbol', [cs, "symbol_1", resolve_symbol]))
self.ws.send(self.generate_message('create_series', [cs, "s1", "s1", "symbol_1", one_session_dict[list(one_session_dict.keys())[0]]['tf'], 500]))
one_session_dict['create_study'] = 'st1'
one_session_dict_list.append(one_session_dict)
self.ws.send(self.generate_message('create_study', [cs,"st1","st1","s1","Script@tv-scripting-101!",{"text":"BNd6/a0yiHhiH9R+wMKfHw==2ATBIYa24E+JJoRT7qmFTVweiQNhpyUoCL340t0Zw9e4V7FMpddEizx8UsrzRK40aI5yfLC9h2AQn+/L6OnZkrc+hHgI5Lb+HnCJdU2PcUMRP72Yj8ZUMUFvr4ziJKBSH2V7lmkE47ZJCalmdZX8R5cRmqZcpyJKP6aN/5fuFf6Qv5NSIKtEVSLlJD6CevC+/jLANErYE6dpkCaOs/O0qB2f7FxqOHp9NPztLC3HsJKFXpJ9/rDjBeIrk1I00XSXY5rg/kvU8bIASzgXnT5p3+bKjqViOwrygx/Deu1bRfRbxJNEr8KJACnoKqIQfKCOxTcre1L6icSIBxntdRLe39VCWAhtJ6iDqBSWvQKdO8ASsvGcHDTNS6y19uRvK7BIovux/K0pMFzW0g==","pineId":"STD;MACD","pineVersion":"13.0","in_6":{"v":"","f":True,"t":"resolution"},"in_0":{"v":12,"f":True,"t":"integer"},"in_1":{"v":26,"f":True,"t":"integer"},"in_2":{"v":"close","f":True,"t":"source"},"in_3":{"v":9,"f":True,"t":"integer"},"in_4":{"v":False,"f":True,"t":"bool"},"in_5":{"v":False,"f":True,"t":"bool"}}]))
return one_session_dict_list
def generate_message(self, m, p):
params = json.dumps({"m":m,"p":p}, separators=(',', ':'))
package_symbols = 'm'
data = [package_symbols, str(len(params)), package_symbols, params]
return ''.join(data)
def on_message(self, message):
if re.search('m\d+m~~h\d+', message):
self.ws.send(message)
print(message)
mess_split = [mess for mess in message.split('~', -1)][4::4]
for one_message in mess_split:
if one_message:
if 'st1' in ujson.loads(one_message)['p'][1]:
macd_dict = {}
macd_dict['cs_session'] = ujson.loads(one_message)['p'][0]
macd = ujson.loads(one_message)['p'][1]['st1']['st'][0]['v'][1]
macd_dict['macd'] = macd
macd_signal = ujson.loads(one_message)['p'][1]['st1']['st'][0]['v'][3]
macd_dict['macd_signal'] = macd_signal
macd_dict['tf'] = [ses['MOEX:SIH2021']['tf'] for ses in one_session_dict_list if macd_dict['cs_session'] == ses['MOEX:SIH2021']['cs']][0]
def create_dict_of_symbol(self):
symbols = set(re.sub(' +', '', "MOEX:SIH2021").split(','))
result = []
for tf in ["1","3","5","10","15","30","45","60","120","180","240","1D"]:
for symbol in symbols:
result_dict = {}
result_dict[symbol] = {'cs': self.generate_session('cs'), 'qs': self.generate_session('qs'),'tf':tf}
result.append(result_dict)
return result
def generate_session(self,type_session, length=12):
letters = string.ascii_letters
return type_session + '' + ''.join(random.sample(letters, length))
url_ws = wss://data.tradingview.com/socket.io/websocket
def get_auth_token(self):
data = {"username": "your_tradingview_username", "password": "your_tradingview_password", "remember": "on"}
headers = {'Referer': 'https://ru.tradingview.com', 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.198 Safari/537.36','authority': 'ru.tradingview.com','x-language': 'ru'}
response = requests.post(url="https://www.tradingview.com/accounts/signin/", data=data, headers=headers)
auth_token = response.json()['user']['auth_token']
return auth_token
def on_open(self):
self.ws.send(self.generate_message('set_auth_token', [self.get_auth_token()]))
dict_session = self.create_dict_of_symbol()
one_session_dict_list = []
for one_session_dict in dict_session:
cs = one_session_dict[list(one_session_dict.keys())[0]]['cs']
self.ws.send(self.generate_message('chart_create_session', [cs, ""]))
resolve_symbol = "={"symbol":"" + list(one_session_dict.keys())[0] + "","adjustment":"splits"}"
self.ws.send(self.generate_message('resolve_symbol', [cs, "symbol_1", resolve_symbol]))
self.ws.send(self.generate_message('create_series', [cs, "s1", "s1", "symbol_1", one_session_dict[list(one_session_dict.keys())[0]]['tf'], 500]))
one_session_dict['create_study'] = 'st1'
one_session_dict_list.append(one_session_dict)
self.ws.send(self.generate_message('create_study', [cs,"st1","st1","s1","Script@tv-scripting-101!",{"text":"BNd6/a0yiHhiH9R+wMKfHw==2ATBIYa24E+JJoRT7qmFTVweiQNhpyUoCL340t0Zw9e4V7FMpddEizx8UsrzRK40aI5yfLC9h2AQn+/L6OnZkrc+hHgI5Lb+HnCJdU2PcUMRP72Yj8ZUMUFvr4ziJKBSH2V7lmkE47ZJCalmdZX8R5cRmqZcpyJKP6aN/5fuFf6Qv5NSIKtEVSLlJD6CevC+/jLANErYE6dpkCaOs/O0qB2f7FxqOHp9NPztLC3HsJKFXpJ9/rDjBeIrk1I00XSXY5rg/kvU8bIASzgXnT5p3+bKjqViOwrygx/Deu1bRfRbxJNEr8KJACnoKqIQfKCOxTcre1L6icSIBxntdRLe39VCWAhtJ6iDqBSWvQKdO8ASsvGcHDTNS6y19uRvK7BIovux/K0pMFzW0g==","pineId":"STD;MACD","pineVersion":"13.0","in_6":{"v":"","f":True,"t":"resolution"},"in_0":{"v":12,"f":True,"t":"integer"},"in_1":{"v":26,"f":True,"t":"integer"},"in_2":{"v":"close","f":True,"t":"source"},"in_3":{"v":9,"f":True,"t":"integer"},"in_4":{"v":False,"f":True,"t":"bool"},"in_5":{"v":False,"f":True,"t":"bool"}}]))
return one_session_dict_list
def generate_message(self, m, p):
params = json.dumps({"m":m,"p":p}, separators=(',', ':'))
package_symbols = '
m'data = [package_symbols, str(len(params)), package_symbols, params]
return ''.join(data)
def on_message(self, message):
if re.search('
m\d+m~~h\d+', message):self.ws.send(message)
print(message)
mess_split = [mess for mess in message.split('~', -1)][4::4]
for one_message in mess_split:
if one_message:
if 'st1' in ujson.loads(one_message)['p'][1]:
macd_dict = {}
macd_dict['cs_session'] = ujson.loads(one_message)['p'][0]
macd = ujson.loads(one_message)['p'][1]['st1']['st'][0]['v'][1]
macd_dict['macd'] = macd
macd_signal = ujson.loads(one_message)['p'][1]['st1']['st'][0]['v'][3]
macd_dict['macd_signal'] = macd_signal
macd_dict['tf'] = [ses['MOEX:SIH2021']['tf'] for ses in one_session_dict_list if macd_dict['cs_session'] == ses['MOEX:SIH2021']['cs']][0]
def create_dict_of_symbol(self):
symbols = set(re.sub(' +', '', "MOEX:SIH2021").split(','))
result = []
for tf in ["1","3","5","10","15","30","45","60","120","180","240","1D"]:
for symbol in symbols:
result_dict = {}
result_dict[symbol] = {'cs': self.generate_session('cs'), 'qs': self.generate_session('qs'),'tf':tf}
result.append(result_dict)
return result
def generate_session(self,type_session, length=12):
letters = string.ascii_letters
return type_session + '' + ''.join(random.sample(letters, length))