kuznetsov-m · October 15, 2021 12:04 · Oct 15, 2021 · Oct 15, 2021 · Oct 15, 2021
diff --git a/sber_price_monitor.py b/sber_price_monitor.py
@@ -51,6 +51,25 @@ def _load_candels_from_moex():
 
         return result_df
 
+    # row - строка из dataframe
+    def _build_message(self, row) -> str:
+        # data = json.loads(df_final.iloc[-1:].to_json(orient='records', date_format = 'iso'))[0]
+        # text = json.dumps(data, indent=0)[2:][:-2]
+        # return text
+
+        end = row['end']
+        close = row['close']
+        step = row['step']
+        step_prev_1 = row['step_prev_1']
+
+        simvol = '⏹'
+        if step > 0:
+            simvol = '↗️'
+        elif step < 0:
+            simvol = '↘️'
+
+        return f'{end.to_pydatetime().strftime("%H:%M:%S %d.%m.%Y %Z")}\n{close} ({simvol} {step})'
+
     def loop(self):
         timeout = 60
 
@@ -120,17 +139,18 @@ def loop(self):
                     # df_final.to_csv('df_final.csv', sep='\t', encoding='utf-8')
                     # # telegram_bot.send_document_to_user('df_final.csv')
                     # print(f'{__name__}: df_final.csv was sent')
-                    ############ 
+                    ############
+
+                    row = df_final.iloc[-1]
+
+                    text = _build_message(row)
 
-                    data = json.loads(df_final.iloc[-1:].to_json(orient='records', date_format = 'iso'))[0]
-                    text = json.dumps(data, indent=0)[2:][:-2]
-                    print(text)
-                    print(f'{__name__}: post text')
+                    print(f'{__name__}: post publication')
                     telegram_client.post_message_to_channel(
                         message=text,
                         channel_url=self._channel_url
                     )
-                    print(f'{__name__}: text has been posted')
+                    print(f'{__name__}: post successfully published')
 
                     last_send_datetime = df_final.iloc[-1]['end']
 
@@ -139,10 +159,10 @@ def loop(self):
     def run(self):
         while True:            
             try:
-                telegram_client.post_message_to_channel(
-                    message='Monitor restarted',
-                    channel_url=self._channel_url
-                )
+                # telegram_client.post_message_to_channel(
+                #     message='Monitor restarted',
+                #     channel_url=self._channel_url
+                # )
                 self.loop()
             except Exception as e:
                 print(f'{__name__}: EXCEPTION: {str(e)}')

diff --git a/sber_price_monitor.py b/sber_price_monitor.py
@@ -63,7 +63,7 @@ def loop(self):
         while True:
             print(f'{__name__}: loop')
 
-            df = db_utils.last_rows_from_sql_table(table_name=models.SberCandels.__tablename__, rows=10000)
+            df = db_utils.last_rows_from_sql_table(table_name=models.SberCandels.__tablename__, rows=2000)
             print(f'{__name__}: df rows count ={len(df.index)}')
 
             dt_now = datetime.combine(datetime.utcnow().date(), datetime.min.time())
@@ -111,9 +111,17 @@ def loop(self):
                 df_final['step_prev_1'] = df_final['step'].shift(1).fillna(0)
                 df_final = df_final.loc[df_final['step'] != df_final['step_prev_1']]
 
+                print(f'{__name__}: df_final rows count ={len(df_final.index)}')
                 print(df_final)
 
                 if not df_final.empty and last_send_datetime != df_final.iloc[-1]['end']:
+                    # debug code block
+                    # print(f'{__name__}: df_final\n{df_final}')
+                    # df_final.to_csv('df_final.csv', sep='\t', encoding='utf-8')
+                    # # telegram_bot.send_document_to_user('df_final.csv')
+                    # print(f'{__name__}: df_final.csv was sent')
+                    ############ 
+
                     data = json.loads(df_final.iloc[-1:].to_json(orient='records', date_format = 'iso'))[0]
                     text = json.dumps(data, indent=0)[2:][:-2]
                     print(text)
@@ -131,6 +139,10 @@ def loop(self):
     def run(self):
         while True:            
             try:
+                telegram_client.post_message_to_channel(
+                    message='Monitor restarted',
+                    channel_url=self._channel_url
+                )
                 self.loop()
             except Exception as e:
                 print(f'{__name__}: EXCEPTION: {str(e)}')

diff --git a/sber_price_monitor.py b/sber_price_monitor.py
@@ -0,0 +1,139 @@
+import pandas as pd
+import numpy as np
+from datetime import datetime
+
+import json
+import time
+from dateutil import tz
+import models
+import db_utils
+import telegram_client
+
+class SberPriceMonitor():
+    _channel_url = 'https://t.me/sber_price_monitor'
+
+    def _load_candels_from_moex():
+        ticker = 'SBER'
+        from_ = datetime.now().strftime('%Y-%m-%d')
+        till = from_
+        interval = 1
+
+        # отладка
+        # from_ = '2021-10-10'
+        # till = '2021-10-13'
+
+        result_df = pd.DataFrame()
+
+        for start in range(0, 10000000000, 500):
+            query = f'http://iss.moex.com/iss/engines/stock/markets/shares/securities/{ticker}/candles.csv?from={from_}&till={till}&interval={interval}&start={start}'
+
+            df = None
+            try_count = 1
+            timeout = 2
+
+            while df is None:
+                try:
+                    content = requests.get(query, timeout=timeout).content
+                    df = pd.read_csv(
+                        io.StringIO(content.decode('utf-8')),
+                        sep=';',
+                        header=1,
+                        parse_dates=['begin','end']
+                    )
+                except Exception as e: 
+                    print(f'{__name__} Error: Try count: {try_count}. Timeout: {timeout}. Error message: {str(e)}')
+                    timeout = min(timeout * 2, 20)
+
+            if df.shape[0]:
+                result_df = result_df.append(df, ignore_index=True)
+            else:
+                break
+
+        return result_df
+
+    def loop(self):
+        timeout = 60
+
+        pd.set_option('display.max_rows', None)
+
+        step = 0.01
+
+        last_send_datetime = None
+
+        while True:
+            print(f'{__name__}: loop')
+
+            df = db_utils.last_rows_from_sql_table(table_name=models.SberCandels.__tablename__, rows=10000)
+            print(f'{__name__}: df rows count ={len(df.index)}')
+
+            dt_now = datetime.combine(datetime.utcnow().date(), datetime.min.time())
+            # dt_now = dt_now.astimezone(tz.UTC)
+
+            # отладка
+            # date_time_obj = datetime.strptime('14.10.21 14:55:19', '%d.%m.%y %H:%M:%S')
+            # dt_now = datetime.combine(date_time_obj.date(), datetime.min.time())
+
+            mask = (df['begin'].dt.tz_localize(None) >= np.datetime64(dt_now))
+            df = df.loc[mask]
+            print(f'{__name__}: df rows count after filter ={len(df.index)}')
+
+            # Оставляем только сегодняшние строки
+            # df.loc[df['begin'].dt.date == pd.to_datetime('today').normalize()]
+            # print(df.iloc()[:1])
+
+            if not df.empty:
+                df.drop(columns=['begin','high','low','value','volume'], inplace=True)
+                df['end'] += pd.to_timedelta(1, unit='s')
+
+                df['frac'] = (df['close'] / df['open'].iloc[0]) - 1
+                df['step'] = np.nan
+
+                for val in range(0, 20):
+                    df.loc[(df['frac'] >= val*step) & (df['frac'] < (val+1)*step), 'step'] = val*step
+                for val in range(-20, 0):
+                    df.loc[(df['frac'] >= val*step) & (df['frac'] < (val+1)*step), 'step'] = (val+1)*step
+
+                df['step_prev_1'] = df['step'].shift(1).fillna(0)
+                df['step_prev_2'] = df['step'].shift(2).fillna(0)
+                df['step_prev_3'] = df['step'].shift(3).fillna(0)
+                df['step_prev_4'] = df['step'].shift(4).fillna(0)
+                df['step_prev_5'] = df['step'].shift(5).fillna(0)
+
+                df_final = df.loc[
+                    (df['step'] == df['step_prev_1']) &
+                    (df['step'] == df['step_prev_2']) &
+                    (df['step'] == df['step_prev_3']) &
+                    (df['step'] == df['step_prev_4']) &
+                    (df['step'] != df['step_prev_5']),
+                    ['end','close','frac','step']
+                ]
+
+                df_final['step_prev_1'] = df_final['step'].shift(1).fillna(0)
+                df_final = df_final.loc[df_final['step'] != df_final['step_prev_1']]
+
+                print(df_final)
+
+                if not df_final.empty and last_send_datetime != df_final.iloc[-1]['end']:
+                    data = json.loads(df_final.iloc[-1:].to_json(orient='records', date_format = 'iso'))[0]
+                    text = json.dumps(data, indent=0)[2:][:-2]
+                    print(text)
+                    print(f'{__name__}: post text')
+                    telegram_client.post_message_to_channel(
+                        message=text,
+                        channel_url=self._channel_url
+                    )
+                    print(f'{__name__}: text has been posted')
+
+                    last_send_datetime = df_final.iloc[-1]['end']
+
+            time.sleep(timeout)
+
+    def run(self):
+        while True:            
+            try:
+                self.loop()
+            except Exception as e:
+                print(f'{__name__}: EXCEPTION: {str(e)}')
+
+monitor = SberPriceMonitor()
+monitor.run()