enhanced async, and notification for debugging

backgroundTask.py

@@ -1,5 +1,9 @@
+# import sys
+# sys.path.append('/Users/lamonkey/Desktop/portfolio_management')
 import sys
-sys.path.append('/Users/lamonkey/Desktop/portfolio_management')
+# add pipeline to path
+import os
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__))))
 from pipeline import daily_update
 import panel as pn
 from datetime import timedelta

index_page.py

@@ -14,37 +14,42 @@ engine = create_engine(db_url)
 
 analytic_p = db.get_portfolio_analytic_df()
 analytic_b = db.get_benchmark_analytic_df()
+if len(analytic_p) == 0:
+    app = [
+        pn.pane.Str('No portfolio data found, please update portfolio first')]
 
-stock_overview = appComponents.BestAndWorstStocks(
-    analytic_df=analytic_p)
-composation_card = appComponents.PortfolioComposationCard(
-    analytic_p)
-monthly_return_card = appComponents.HistReturnCard(
-    calculated_p_stock=analytic_p, calculated_b_stock=analytic_b)
-total_return_card = appComponents.TotalReturnCard(name='Range',
-                                                  b_stock_df=analytic_b,
-                                                  p_stock_df=analytic_p,
-                                                  value=(0, 20))
-drawdown_card = appComponents.DrawDownCard(
-    calculated_p_stock=analytic_p)
-
-top_header = appComponents.TopHeader(
-    eval_df=analytic_p,
-)
 
-template = pn.template.FastListTemplate(
-    title="Portfolio一览",
-    # sidebar=[freq, phase],
-)
-template.sidebar.append(SideNavBar())
-# template.main.extend([drawdown_card, stock_overview, composation_card, monthly_return_card, total_return_card])
-template.main.extend(
-    [pn.Row(top_header),
-     pn.Row(
+else:
+
+    stock_overview = appComponents.BestAndWorstStocks(
+        analytic_df=analytic_p)
+    composation_card = appComponents.PortfolioComposationCard(
+        analytic_p)
+    monthly_return_card = appComponents.HistReturnCard(
+        calculated_p_stock=analytic_p, calculated_b_stock=analytic_b)
+    total_return_card = appComponents.TotalReturnCard(name='Range',
+                                                      b_stock_df=analytic_b,
+                                                      p_stock_df=analytic_p,
+                                                      value=(0, 20))
+    drawdown_card = appComponents.DrawDownCard(
+        calculated_p_stock=analytic_p)
+
+    top_header = appComponents.TopHeader(
+        eval_df=analytic_p,
+    )
+    app = [pn.Row(top_header),
+           pn.Row(
         pn.Column(monthly_return_card, stock_overview,
                   width=500, margin=(10, 10, 10, 10)),
         pn.Column(total_return_card, drawdown_card, margin=(10, 10, 10, 10)),
         pn.Column(composation_card, margin=(10, 10, 10, 10)),
     )]
+
+template = pn.template.FastListTemplate(
+    title="Portfolio一览",
+    # sidebar=[freq, phase],
 )
+template.sidebar.append(SideNavBar())
+# template.main.extend([drawdown_card, stock_overview, composation_card, monthly_return_card, total_return_card])
+template.main.extend(app)
 template.servable()

pipeline.py

@@ -9,11 +9,13 @@ import table_schema as ts
 import db_operation as db
 from log import Log
 import processing
+from tornado import gen
 # import settings
 # fetch new stock price
 stock_price_stream = Stream()
-event = Stream()
-
+event = Stream(asynchronous=True)
+# display notification to 
+notification_queue = Stream()
 # log
 log = Log('instance/log.json')
 # save stock price to db
@@ -27,71 +29,6 @@ log = Log('instance/log.json')
 # run using  --setup
 db_url = 'sqlite:///instance/local.db'
 
-
-def create_portfolio_profile_df(stocks: list[dict]):
-    profile_df = pd.DataFrame(stocks)
-    profile_df = add_details_to_stock_df(profile_df)
-
-    # check if there is duplicate ticker
-    if profile_df.ticker.duplicated().any():
-        raise Exception(
-            'VALIDATION ERROR: cannot have duplicate ticker with the same date')
-
-    return profile_df
-
-
-def need_to_update(table_name: str, freq: dt.datetime):
-    '''check table with table_name need to update
-    Return
-    ------
-    None if no need to update
-    (start_date, end_date, freq) if need to update
-    '''
-    with create_engine(db_url).connect() as conn:
-        max_date = conn.execute(
-            text(f"SELECT MAX(date) FROM {table_name}")).fetchone()[0]
-        max_date = utils.convert_string_to_datetime(max_date)
-        current_time = utils.time_in_beijing()
-        if current_time - max_date > freq:
-            return (max_date + freq, current_time, freq)
-        else:
-            return None
-
-
-def need_to_forward_fill_stock_price():
-    '''
-    check if need to pull new stock price from jq
-
-    RETURN
-    ------
-    (min_date, max_date) : if update is needed
-       the start and end date need to fetch new stock price
-    None if no need to fetch new stock price
-
-    '''
-    # TODO refactor to db
-    # get min date from portfolio_profile
-    with create_engine(db_url).connect() as conn:
-        table_name = 'portfolio_profile'
-        query = f"SELECT DISTINCT date FROM {table_name} ORDER BY date ASC LIMIT 1"
-        df = pd.read_sql(query, con=conn)
-        df.date = pd.to_datetime(df.date)
-        min_date = df.date[0]
-    # TODO refactor to db
-    # compare to min date from stocks_price
-    with create_engine(db_url).connect() as conn:
-        table_name = 'stocks_price'
-        query = f"SELECT DISTINCT time FROM {table_name} ORDER BY time ASC LIMIT 1"
-        df = pd.read_sql(query, con=conn)
-        df.time = pd.to_datetime(df.time)
-
-    # return
-    if min_date <= df.time[0]:
-        return (min_date, df.time[0] - dt.timedelta(days=1))
-    else:
-        return None
-
-
 def get_most_recent_profile(type):
     table_name = 'benchmark_profile' if type == 'benchmark' else 'portfolio_profile'
     query = f"SELECT * FROM {table_name} WHERE date = (SELECT MAX(date) FROM {table_name})"
@@ -193,21 +130,15 @@ def update_portfolio_profile_to_db(portfolio_df):
 
     if (_validate_schema(portfolio_df, ts.PORTFOLIO_TABLE_SCHEMA)):
         raise ValueError(
-            'portfoliijuo_df has different schema than PORTFOLIO_DB_SCHEMA')
-
-    with create_engine(db_url).connect() as conn:
-        print("updating profile to db")
-        try:
+            'portfolio_ef has different schema than PORTFOLIO_DB_SCHEMA')
+    try:
+        with create_engine(db_url).connect() as conn:
             portfolio_df[ts.PORTFOLIO_TABLE_SCHEMA.keys()].to_sql(
-                ts.PORTFOLIO_TABLE, con=conn, if_exists='append', index=False)
+                ts.PORTFOLIO_TABLE, con=conn, if_exists='replace', index=False)
             
-            event.emit('update_portfolio')
-            # handle_portfolio_update()
-           
-            return True
-        except:
-            return False
-    
+    except Exception as e:
+        print(e)
+        raise e
 
 
 def right_fill_stock_price():
@@ -501,11 +432,12 @@ async def run():
             # print(stock_df)
             stock_price_stream.emit(stock_df)
 
-
+@gen.coroutine
 def handle_event(e):
     if e == "update_portfolio":
         print("handling portfolio update")
         handle_portfolio_update()
         print("done handling portfolio update")
 
-event.sink(handle_event)
+
+event.sink(handle_event)

portfolioEditingPage.py

@@ -1,5 +1,6 @@
 # %%
 # load portfolio
+import time
 import panel as pn
 from utils import create_stocks_entry_from_excel, style_number, create_share_changes_report
 import datetime as dt
@@ -8,10 +9,11 @@ from utils import time_in_beijing
 import api
 import pandas as pd
 from sqlalchemy import create_engine
-from pipeline import update_portfolio_profile_to_db
+
 import table_schema
-import pipeline
+
 import db_operation as db
+import pipeline
 from sidebar import SideNavBar
 db_url = 'sqlite:///instance/local.db'
 pn.extension()
@@ -26,27 +28,88 @@ pn.extension(notifications=True)
 MIN_COMPONENT_WIDTH = 375
 MAX_COMPONENT_WIDTH = 600
 
-# %%
+def create_portfolio_stream_entry(stocks, portfolio_df):
+    
+    # create entry with ticker, date and details
+    stream_entry = pd.DataFrame(stocks)
+
+    # if have duplicate ticker raise error
+    if stream_entry.ticker.duplicated().any():
+        raise Exception('VALIDATION_ERROR: plase remove duplicate ticker')
+    
+    # raise error if portfolio already have same ticker with same date
+    date = stream_entry.date[0]
+    selected_df = portfolio_df[portfolio_df.date == date]
+    tickers = stream_entry.ticker.tolist()
+    filter_out_ticker = selected_df[selected_df.ticker.isin(tickers)].ticker
+
+    if len(filter_out_ticker) > 0:
+        raise Exception(f'VALIDATION_ERROR: {" ".join(filter_out_ticker)}在{date}已存在,请先删除再添加')
+
+
+    stream_entry = pipeline.add_details_to_stock_df(stream_entry)
+
+    # calculate share changes, use tmp_df to save intermediate result
+    tmp_df = pd.concat([stream_entry, portfolio_df], ignore_index=True, join='outer')
+    tmp_df.sort_values(by='date', inplace=True)
+    tmp_df['share_changes'] = tmp_df.groupby('ticker').shares.diff()
+
+    # for ticker previous not existing use shares as share_changes
+    tmp_df.share_changes = tmp_df.share_changes.fillna(tmp_df.shares)
+
+    # add share_chagnes back to stream_entry
+    stream_entry = stream_entry.merge(
+        tmp_df[['ticker','date','share_changes','change_saved']],
+        on=['ticker','date'],
+        how='left'
+    )
+
+    # indicate not saved
+    stream_entry['change_saved'] = False
+
+    # indicate sync to db
+    stream_entry['sync_to_db'] = True
+
+    # fill empty ave_price with latest closing price 
+    # TODO: for now all ave_price is fetching from api
+    ticker = stream_entry.ticker.tolist()
+    close_price = api.fetch_stocks_price(security=ticker, end_date=date, frequency='minute', count=1)[['ticker','close']]
+    close_price.rename(columns={'close':'ave_price'}, inplace=True)
+    stream_entry = stream_entry.merge(close_price, on='ticker', how='left')
+
+    # calculate cash(mkt_value) and weight
+    stream_entry['cash'] = stream_entry.shares * stream_entry.ave_price
+    stream_entry['weight'] = stream_entry.cash / stream_entry.cash.sum()
+    return stream_entry
+
 
 def notify(func):
     def wrapper(*args, **kwargs):
         try:
-            notification = func(*args, **kwargs)
-            if notification is not None:
-                if notification['type'] == 'success':
-                    pn.state.notifications.success(notification['description'], duration=4000)
-                elif notification['type'] == 'error':
-                    pn.state.notifications.error(notification['description'], duration=4000)
-                elif notification['type'] == 'warning':
-                    pn.state.notifications.warning(notification['description'], duration=4000)
-                elif notification['type'] == 'info':
-                    pn.state.notifications.info(notification['description'], duration=4000)
-                else:
-                    raise Exception('unknow notification type')  
+            notifications = func(*args, **kwargs)
+
+            if notifications is not None:
+                for notification in notifications:
+                    duration = notification.get('duration', 4000)
+                    if notification['type'] == 'success':
+                        pn.state.notifications.success(
+                            notification['description'], duration=duration)
+                    elif notification['type'] == 'error':
+                        pn.state.notifications.error(
+                            notification['description'], duration=duration)
+                    elif notification['type'] == 'warning':
+                        pn.state.notifications.warning(
+                            notification['description'], duration=duration)
+                    elif notification['type'] == 'info':
+                        pn.state.notifications.info(
+                            notification['description'], duration=duration)
+                    else:
+                        raise Exception('unknow notification type')
         except Exception as e:
-            pn.state.notifications.error(str(e), duration=4000)
+            pn.state.notifications.error(str(e), duration=0)
     return wrapper
 
+
 def app():
 
     # load portfolio df
@@ -97,12 +160,12 @@ def app():
     portfolio_tabulator = pn.widgets.Tabulator(p_profile,
                                                layout='fit_columns',
                                                height_policy='max',
-                                               width=1000,
+                                               #    width=1000,
                                                groupby=['date'],
                                                hidden_columns=hidden_column, titles=col_to_titles,
                                                formatters=tabulator_formatters,
                                                editors=bokeh_editors,
-                                               pagination='local',
+                                               pagination='remote',
                                                # page_size=25,
                                                #    frozen_columns=frozen_columns
                                                )
@@ -123,7 +186,9 @@ def app():
                                              hidden_columns=hidden_column,
                                              height_policy='max',
                                              titles=col_to_titles)
-
+    # perform calculation btn
+    force_recalculate_btn = pn.widgets.Button(
+        name='重新计算', button_type='primary', sizing_mode='stretch_width')
     # create component
     new_stock_btn = pn.widgets.Button(
         name='增加新股票', button_type='primary', sizing_mode='stretch_width')
@@ -142,6 +207,7 @@ def app():
         width_policy='max', height_policy='max', scroll=True)
     # floating window row
     floating_windows = pn.Row()
+
     @notify
     def _update_history_tabulator(action, df=None):
         '''handle update history tabulator'''
@@ -165,18 +231,21 @@ def app():
                 history_tabulator.patch({
                     'change_saved': [(index[0], False)]
                 }, as_index=True)
-            
-            return {'type':'warning','description':'添加成功请保存'}
+
+            yield {'type': 'warning', 'description': '添加成功请保存'}
         # hanlde editing portoflio tabulator
         elif action == 'edit':
             # mark synced_to_db to false when entry is edited
             date = df
             index = history_tabulator.value[history_tabulator.value.date == date].index.to_list(
             )
+            # check if all change saved
+            all_saved = all(
+                portfolio_tabulator.value[portfolio_tabulator.value.date == date]['change_saved'])
             history_tabulator.patch({
-                'change_saved': [(index[0], False)]
+                'change_saved': [(index[0], all_saved)]
             }, as_index=True)
-            return {'type':'warning','description':'修改成功请保存'}
+            yield {'type': 'warning', 'description': '修改成功请保存'}
         # handle sync to db
         elif action == 'sync':
             # patch all synced_to_db to true
@@ -188,12 +257,13 @@ def app():
             history_tabulator.patch({
                 'change_saved': [(index, True) for index in indices]
             }, as_index=True)
-            return {'type':'success','description':'同步成功以更新'}
+            yield {'type': 'success', 'description': '同步成功以更新'}
+
     @notify
     def delete_stock(row):
         '''delete a stock entry'''
         stock_column.remove(row)
-        return {'type':'success','description':'删除成功'}
+        yield {'type': 'success', 'description': '删除成功'}
 
     def create_new_stock_entry(ticker=None, shares=0, ave_price=0.0, disable_ticker=True):
         '''create a new new stock entry'''
@@ -214,15 +284,10 @@ def app():
             start=0,
             sizing_mode='stretch_width')
 
-        mean_price_input = pn.widgets.FloatInput(
-            name='平均成本',
-            value=ave_price, step=0.01, start=0, sizing_mode='stretch_width')
-
         row = pn.Row(
             delete_btn,
             ticker_selector,
             share_input,
-            mean_price_input,
             width_policy='max',
         )
         delete_btn.on_click(lambda _, row=row: delete_stock(row))
@@ -250,48 +315,36 @@ def app():
         stock_column.clear()
         stock_column.extend(stock_entries)
 
+    @notify
     def update_profile_tabulator(e):
         '''add all stocks entry to ui'''
-        new_entry = [dict(ticker=row[1].value,
-                          shares=row[2].value,
-                          ave_price=row[3].value,
-                          date=datetime_picker.value) for row in stock_column]
-
-        if len(new_entry) == 0:
-            print("no entry added")
-            return
-
-        new_profile = pipeline.create_portfolio_profile_df(new_entry)
-        # calculate share changes
-        tmp_profile = pd.concat([p_profile, new_profile], ignore_index=True)
-        tmp_profile.sort_values(by='date', inplace=True)
-        tmp_profile['share_changes'] = tmp_profile.groupby('ticker')[
-            'shares'].diff()
-        tmp_profile['share_changes'] = tmp_profile['share_changes'].fillna(
-            tmp_profile['shares'])
-        new_profile = new_profile.merge(tmp_profile[[
-                                        'ticker', 'date', 'share_changes', 'change_saved']], on=['ticker', 'date'], how='left')
-        # fill emtpy change_saved to False
-        new_profile['change_saved'] = new_profile['change_saved'].fillna(False)
-        new_profile['sync_to_db'] = True
-        # calculate cash and weight
-        new_profile['cash'] = new_profile.shares * new_profile.ave_price
-        new_profile['weight'] = new_profile.cash / new_profile.cash.sum()
-
-        # update history tabulator
-        _update_history_tabulator('append', new_profile)
-        _stream_to_portfolio_tabulator(new_profile)
+        # TODO: make this idempotent 
+        new_entries = [dict(ticker=row[1].value,
+                            shares=row[2].value,
+                            date=datetime_picker.value) for row in stock_column]
         
+        try:
+            new_profile = create_portfolio_stream_entry(new_entries, portfolio_tabulator.value)
+            # update history tabulator
+            _update_history_tabulator('append', new_profile)
+            _stream_to_portfolio_tabulator(new_profile)
+            yield {'type': 'success', 'description': f'已添加{len(new_entries)}条新股票,请保存'}
+        except Exception as e:
+            raise Exception(e)
 
     def add_new_stock(e):
         row = create_new_stock_entry()
         stock_column.append(row)
 
+    @notify
     def _stream_to_portfolio_tabulator(entry):
+        # not using stream because it will cause index mismatch
         if len(portfolio_tabulator.value) == 0:
             portfolio_tabulator.value = entry
+
         else:
             portfolio_tabulator.stream(entry, follow=True)
+            yield {'type': 'success', 'description': f'添加{len(entry)}条股票'}
 
     def handle_click_on_history_tabulator(e):
         '''handle click click on history tabulator'''
@@ -311,25 +364,57 @@ def app():
         new_portfolio = portfolio_tabulator.value
         # only update selected row to db
         selected_portfolio = new_portfolio[new_portfolio['sync_to_db']]
-        successed = update_portfolio_profile_to_db(selected_portfolio)
+        try:
+            pipeline.update_portfolio_profile_to_db(selected_portfolio)
+        except Exception as e:
+            raise Exception(f'同步到数据库失败,错误信息:{e}')
         # update history tabulator and portfolio tabulator
-        if successed:
-            # mark changes as saved
-            indices = selected_portfolio[~selected_portfolio['change_saved']].index.to_list(
-            )
-            portfolio_tabulator.patch({
-                'change_saved': [(index, True) for index in indices]
-            }, as_index=True)
-            _update_history_tabulator('sync')
-            return {'type': 'success', 'description': '保存成功'}
-        else:
-            return {'type':'error','description':'保存失败'}
+
+        # mark changes as saved
+        indices = selected_portfolio[~selected_portfolio['change_saved']].index.to_list(
+        )
+        portfolio_tabulator.patch({
+            'change_saved': [(index, True) for index in indices]
+        }, as_index=True)
+        _update_history_tabulator('sync')
+        yield {'type': 'success', 'description': '保存成功'}
+
     def handle_edit_portfolio_tabulator(e):
         date = portfolio_tabulator.value.iloc[e.row]['date']
         _update_history_tabulator(df=date, action='edit')
-        print(date)
 
-    # %%
+    def hanlde_edit_history_tabulator(e):
+        # toggle sync on all entry on a date
+        if e.column == 'sync_to_db':
+            date = history_tabulator.value.iloc[e.row]['date']
+            # index of all entry on portfolio tabulator
+            indices = portfolio_tabulator.value[portfolio_tabulator.value.date.between(
+                date.replace(microsecond=0), date.replace(microsecond=999999))].index.to_list()
+            # patch all indices on sync_to_db to e.value
+            portfolio_tabulator.patch({
+                'sync_to_db': [(index, e.value) for index in indices]
+            }, as_index=True)
+
+    @notify
+    def handle_force_recalculation(e):
+        try:
+            yield {'type': 'info', 'description': "开始重新计算可能会花费1分钟以上", 'duration': 0}
+            # fill missing benchmark profile
+            yield {'type': 'info', 'description': "正在获取benchmark数据", 'duration': 0}
+            pipeline.left_fill_benchmark_profile()
+            # fill missing stock price
+            yield {'type': 'info', 'description': "正在更新股票数据", 'duration': 0}
+            pipeline.left_fill_stocks_price()
+
+            # recalculate
+            yield {'type': 'info', 'description': "正在重新计算权重", 'duration': 0}
+            pipeline.batch_processing()
+
+            yield {'type': 'info', 'description': '完成✅', 'duration': 0}
+
+        except Exception as e:
+            raise Exception(f'重新计算失败,错误信息:{e}')
+
     # register event handler
     upload_to_db_btn.on_click(handle_sync_to_db)
     preview_btn.on_click(update_profile_tabulator)
@@ -337,41 +422,43 @@ def app():
     history_tabulator.on_click(
         handle_click_on_history_tabulator
     )
+    force_recalculate_btn.on_click(handle_force_recalculation)
     portfolio_tabulator.on_edit(handle_edit_portfolio_tabulator)
-
-    # %%
+    history_tabulator.on_edit(hanlde_edit_history_tabulator)
     # create handler component to add to panel so can be listened to
     upload_xlsx_handler = pn.bind(update_stock_column, file_input)
 
-    # %%
     # layout
 
     editor_widget = pn.Column(floating_windows, datetime_picker, upload_to_db_btn, new_stock_btn,
-                              preview_btn, file_input, pn.widgets.TooltipIcon(
+                              preview_btn, force_recalculate_btn, file_input, pn.widgets.TooltipIcon(
                                   value="用于更新修改持仓信息,默认股票为最近持仓，默认时间为目前北京时间，点击增加新股票按钮,输入股票代码和持仓选择日期(北京时间),点击预览,确认无误后点击保存到数据库。或者直接拖拽excel文件到下方上传按钮"),
                               stock_column, width=MIN_COMPONENT_WIDTH, height_policy='max')
     # tooltip
     toolTip2 = pn.widgets.TooltipIcon(
         value="持仓总结,每一行的已同步到数据库代表所做更改是否已同步到数据库,点击保存到数据库将上传所有更改。点击右侧📋按钮查看详细持仓变化报告")
 
-    return pn.Row(
-        pn.layout.HSpacer(),
-        editor_widget,
-        pn.Spacer(width=10),
-        history_tabulator,
-        pn.Spacer(width=10),
-        portfolio_tabulator,
-        pn.Spacer(width=10),
-        upload_xlsx_handler,
-        pn.layout.HSpacer(),
-        height=1500,
-        # width_policy='max', height_policy='max')
-        # sizing_mode='stretch_both',
-    )
+    return pn.Column(
+
+        pn.Row(
+            pn.layout.HSpacer(),
+            editor_widget,
+            pn.Spacer(width=10),
+            history_tabulator,
+            pn.Spacer(width=10),
+            portfolio_tabulator,
+            pn.Spacer(width=10),
+            upload_xlsx_handler,
+            pn.layout.HSpacer(),
+            height=1500,
+            # width_policy='max', height_policy='max')
+            # sizing_mode='stretch_both',
+        ))
 
 
 # app
-template = pn.template.FastListTemplate(title='portfolio编辑')
+template = pn.template.FastListTemplate(
+    title='portfolio编辑', sidebar_width=200, collapsed_sidebar=True)
 template.sidebar.append(SideNavBar())
 template.main.append(app())
 template.servable()

processing.py

@@ -423,7 +423,13 @@ def create_analytic_df(price_df, profile_df):
     filling information from profile df to stock price df
 
     '''
+    # daily stock price use begin of the date, need to convert profile_df day to begin of the date
+    profile_df['time'] = profile_df['time'].map(
+        lambda x: datetime(x.year, x.month, x.day))
+    
+    # make every time entry the same dimension
     uni_profile_df = _uniformize_time_series(profile_df)
+    
     # TODO handle rename column here
     df = price_df.merge(uni_profile_df, on=['ticker', 'time'], how='outer')
     df.sort_values(by=['ticker', 'time'], inplace=True)