imp left_fill, right_fill and update log

.vscode/settings.json

@@ -0,0 +1,11 @@
+{
+    "[python]": {
+        "editor.defaultFormatter": "ms-python.autopep8"
+        
+    },
+    "python.linting.flake8Args": [
+        "--ignore=E501"
+    ]
+    
+    
+}

Dockerfile

@@ -13,8 +13,8 @@ COPY . /code
 VOLUME /code/instance
 
 
-# CMD ["panel", "serve", "/code/portfolioEditingPage.py", "--address", "0.0.0.0", "--port", "7860", "--allow-websocket-origin", "lamonkey-portfolio-management.hf.space"]
-CMD ["panel", "serve", "/code/portfolioEditingPage.py", "--address", "0.0.0.0", "--port", "7860", "--allow-websocket-origin", "*"]
+CMD ["panel", "serve", "/code/portfolioEditingPage.py", "--address", "0.0.0.0", "--port", "7860", "--allow-websocket-origin", "lamonkey-portfolio-management.hf.space"]
+# CMD ["panel", "serve", "/code/portfolioEditingPage.py", "--address", "0.0.0.0", "--port", "7860", "--allow-websocket-origin", "*"]
 RUN mkdir /.cache
 RUN chmod 777 /.cache
 RUN mkdir .chroma

api.py

@@ -299,9 +299,9 @@ def fetch_benchmark_profile(start_date: datetime, end_date: datetime, delta_time
     update_df = pd.concat(results)
     update_df['ticker'] = update_df.index
     update_df['date'] = pd.to_datetime(update_df['date'])
-    update_df.rename({'date': 'time'}, inplace=True, axis=1)
+    # update_df.rename({'date': 'time'}, inplace=True, axis=1)
     # remove duplicate row
     update_df = update_df.drop_duplicates(
-        subset=['ticker', 'time'], keep='last')
+        subset=['ticker', 'date'], keep='last')
     update_df.reset_index(drop=True, inplace=True)
     return update_df

db_operation.py

@@ -34,12 +34,27 @@ def _validate_schema(df, schema):
     #         return False
     return True
 
-# def append_benchmark_profile(df):
-#     '''append new entry to benchmark profile table'''
-#     with create_engine(db_url).connect() as conn:
-#         df.to_sql(ts.BENCHMARK_PROFILE_TABLE, con=conn, if_exists='append', index=False)
-
 
+def get_all_benchmark_profile():
+    '''return all entries in the benchmark profile table'''
+    return _get_all_row(ts.BENCHMARK_TABLE)
+
+def append_to_benchmark_profile(df):
+    '''append new entry to benchmark profile table'''
+    # handle possible duplication caused by right fill
+    most_recent_dates = get_most_recent_benchmark_profile().date
+    if len(most_recent_dates) > 0:
+        date = most_recent_dates[0]
+        # drop df.date == date
+        df = df[df.date != date]
+        if len(df) != 0:
+            _append_df_to_db(df, ts.BENCHMARK_TABLE, ts.BENCHMARK_TABLE_SCHEMA)
+    else:
+        _append_df_to_db(df, ts.BENCHMARK_TABLE, ts.BENCHMARK_TABLE_SCHEMA)
+
+def get_most_recent_benchmark_profile():
+    '''return the most recent entry in the benchmark profile table'''
+    return _get_most_recent(ts.BENCHMARK_TABLE)
 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})"
@@ -110,9 +125,10 @@ def get_all_stocks():
         all_stocks = pd.read_sql(ts.STOCKS_DETAILS_TABLE, con=conn)
         return all_stocks
 
-def _get_all_row(table_name):
+def _get_all_row(table_name, ts_column='date'):
     with create_engine(db_url).connect() as conn:
         df = pd.read_sql(table_name, con=conn)
+        df[ts_column] = pd.to_datetime(df[ts_column])
         return df
 
 def get_all_stocks_price():

instance/local.db

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:554c882ab3dd2eb175ca2076ebf2458988340f165620dd03f487d3d23e6f13ea
-size 6856704
+oid sha256:b0651e0e8a48a69d61d008bf5ae3262ca38a8968b020e700604841d55f8784a3
+size 8728576

instance/log.json

@@ -0,0 +1,3 @@
+{
+    "daily_update": "2023-08-29 03:02:51"
+}

newBackgroundTask.py

@@ -1,12 +0,0 @@
-import sys
-sys.path.append('/Users/lamonkey/Desktop/risk-monitor-dashboard')
-from pipeline import update
-import panel as pn
-from datetime import timedelta
-
-# pn.state.schedule_task(
-#     'task', run, period=timedelta(seconds=3)
-# )
-
-# update stock price and benchmark profile
-update()

pipeline.py

@@ -1,24 +1,20 @@
-import sys
-# sys.path.append('/Users/lamonkey/Desktop/risk-monitor-dashboard')
-import panel as pn
 import datetime as dt
-import asyncio
-import random
 from sqlalchemy import create_engine, text
 import pandas as pd
 from streamz import Stream
-from datetime import timedelta
-import settings
-import os
 import utils
 import api
-import numpy as np
 import pytz
 import table_schema as ts
 import db_operation as db
+from log import Log
+# import settings
 # fetch new stock price
 stock_price_stream = Stream()
 
+
+# log
+log = Log('instance/log.json')
 # save stock price to db
 # stock_price_stream.sink(save_stock_price)
 # from dask.distributed import Client
@@ -61,7 +57,7 @@ def need_to_update(table_name: str, freq: dt.datetime):
             return None
 
 
-def need_to_fetch_new_stock_price():
+def need_to_forward_fill_stock_price():
     '''
     check if need to pull new stock price from jq
 
@@ -72,6 +68,7 @@ def 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'
@@ -79,7 +76,7 @@ def need_to_fetch_new_stock_price():
         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'
@@ -105,8 +102,8 @@ def get_most_recent_profile(type):
 
 
 def update_stocks_details_to_db():
-    '''create table contain all stocks detail in db
-    will override existing table if exists
+    '''override stocks_details table with new stocks detail
+
     Table Schema
     ------------
     'display_name', 'name', 'start_date', 'end_date', 'type', 'ticker',
@@ -122,31 +119,6 @@ def update_stocks_details_to_db():
                   if_exists='replace', index=False)
 
 
-def fetch_new_stocks_price():
-    '''
-    get a df contain updated stock prices for both benchmark and portfolio,
-    also indicate if the stock is in portfolio and benchmark
-    '''
-    # most recent profiles
-    p_portfolio = get_most_recent_profile('portfolio')
-    p_benchmark = get_most_recent_profile('benchmark')
-    # combine ticker
-    unique_tickers = pd.concat([p_portfolio, p_benchmark])[
-        'ticker'].unique().tolist()
-    # fetch list of stock
-    # TODO: hard code delta time to 1 day
-    start_date = p_portfolio.date[0] + dt.timedelta(days=1)
-    end_date = utils.time_in_beijing()
-    freq = 'daily'
-    stock_df = api.fetch_stocks_price(
-        unique_tickers, start_date, end_date, freq)
-    stock_df['in_portfolio'] = stock_df['ticker'].isin(
-        p_portfolio['ticker'].unique().tolist())
-    stock_df['in_benchmark'] = stock_df['ticker'].isin(
-        p_benchmark['ticker'].unique().tolist())
-    return stock_df
-
-
 def need_to_update_stocks_price(delta_time):
     # convert p_portfolio.date[0] to timezone-aware datetime object
     tz = pytz.timezone('Asia/Shanghai')
@@ -237,9 +209,10 @@ def update_portfolio_profile_to_db(portfolio_df):
             return False
     # TODO trigger recomputation of analysis
 
-def update_daily_stocks_price():
+
+def right_fill_stock_price():
     '''
-    update all stocks price until today. used for fetching new stock price
+    update all stocks price until today.
 
     if no portfolio, terminate without warning
     default start date is the most recent date in portfolio
@@ -260,36 +233,18 @@ def update_daily_stocks_price():
         new_stocks_price = fetch_all_stocks_price_between(start, end)
         db.append_to_stocks_price_table(new_stocks_price)
 
-def update_stock_price():
-    '''get daily stocks price until today'''
-    # most recent profiles
-    p_portfolio = get_most_recent_profile('portfolio')
-    p_benchmark = get_most_recent_profile('benchmark')
-    # combine ticker
-    unique_tickers = pd.concat([p_portfolio, p_benchmark])[
-        'ticker'].unique().tolist()
-    # fetch list of stock
-    # TODO: hard code delta time to 1 day
-    start_date = p_portfolio.date[0] + dt.timedelta(days=1)
-    end_date = utils.time_in_beijing()
-    freq = 'daily'
-    stock_df = api.fetch_stocks_price(
-        unique_tickers, start_date, end_date, freq)
-    stock_df['in_portfolio'] = stock_df['ticker'].isin(
-        p_portfolio['ticker'].unique().tolist())
-    stock_df['in_benchmark'] = stock_df['ticker'].isin(
-        p_benchmark['ticker'].unique().tolist())
-    return stock_df
-
 
 def fetch_all_stocks_price_between(start, end):
     '''
     patch stock price db with all daily stock price within window
     inclusive on both start and end date
+
     Parameters
     ----------
-    window: tuple
-        (start, end) date of the window
+    start : datetime 
+        start date inclusive
+    end: datetime
+        end date inclusive
 
     Returns
     -------
@@ -302,12 +257,144 @@ def fetch_all_stocks_price_between(start, end):
     tickers = selected_stocks.ticker.to_list()
     # fetch stock price and append to db
     stock_price = api.fetch_stocks_price(
-        security=tickers, start_date=start, end_date=end, frequency='daily')
+        security=tickers,
+        start_date=start,
+        end_date=end,
+        frequency='daily',
+        skip_paused=True,)
     # drop where closing price is null
     stock_price.dropna(subset=['close'], inplace=True)
     return stock_price
 
 
+def right_fill_bechmark_profile():
+    '''
+    right fill the benchmark profile table
+
+    fill any missing entries between the most recent date in benchmark profile and today
+    if no benchmark profile, fill from most recent date in portfolio profile to today
+    if no portfolio profile, terminate without warning
+    '''
+    # get most recent date in benchmark profile
+    b_ends = db.get_most_recent_benchmark_profile().date
+    # get todays date
+    today = utils.time_in_beijing()
+    # most recent portfolio dates
+    p_ends = db.get_most_recent_portfolio_profile().date
+    # if portfolio is empty, terminate
+    if len(p_ends) == 0:
+        return
+    # if no benchmark profile, start is the most recent date in benchmark profile, end is today
+    elif len(b_ends) == 0:
+        start = p_ends[0]
+        end = today
+    # start is most recent benchmark, end is today
+    else:
+        start = b_ends[0]
+        end = today
+
+    # fetch and update
+    new_entry = api.fetch_benchmark_profile(start, end)
+    detailed_new_entry = utils.add_details_to_stock_df(new_entry)
+    db.append_to_benchmark_profile(detailed_new_entry)
+
+
+def left_fill_benchmark_profile():
+    '''
+    left fill the benchmark profile table,
+
+    fill any missing entries between the earliest date in portfolio profile and the earliest date in benchmark profile
+    if no portfolio profile, terminate without warning
+    if no benchmark profile, the span would be from the earliest date in portfolio profile to the most recent date in portfolio profile
+    '''
+    # get starttime of benchmark profile
+    b_starts = db.get_oldest_benchmark_profile().date
+    # get starttime of portfolio profile
+    p_starts = db.get_oldest_portfolio_profile().date
+    # if no portfolio profile, terminate
+    if len(p_starts) == 0:
+        return
+    # use start and end date of portfolio profile if no benchmark profile entry
+    elif len(b_starts) == 0:
+        p_start = p_starts[0]
+        b_start = db.get_most_recent_portfolio_profile().date[0]
+    else:
+        b_start = b_starts[0]
+        p_start = p_starts[0]
+    if p_start < b_start:
+        # back fill benchmark profile
+        new_entry = api.fetch_benchmark_profile(p_start, b_start)
+        detailed_new_entry = utils.add_details_to_stock_df(new_entry)
+        # handle duplicate display_name
+        detailed_new_entry.drop(columns=['display_name_x'], inplace=True)
+        detailed_new_entry.rename(
+            columns={'display_name_y': 'display_name'}, inplace=True)
+
+        # append to db
+        db.append_to_benchmark_profile(detailed_new_entry)
+        # return detailed_new_entry
+    # else do nothing
+
+
+def left_fill_stocks_price():
+    '''
+    left fill stock price
+    fill missing entries between the oldest date in portfolio profile and the oldest date in stock price table
+
+    if no portfolio profile, terminate without warning
+    if no stock price table, the span would be from the oldest date in portfolio profile to the most recent date in portfolio profile
+
+
+    '''
+    # get oldest time in portfolio profile
+    p_start = db.get_oldest_portfolio_profile().date
+    # get oldest time in stock price table
+    stock_start = db.get_oldest_stocks_price().time
+    # if no portfolio profile, terminate
+    if len(p_start) == 0:
+        return
+    # no stock price, span the entire portfolio profile
+    elif len(stock_start) == 0:
+        start = p_start[0]
+        end = db.get_most_recent_portfolio_profile().date[0]
+    else:
+        start = p_start[0]
+        end = stock_start[0]
+    
+    if start < end:
+        # fetch and update
+        new_entry = fetch_all_stocks_price_between(start, end)
+        db.append_to_stocks_price_table(new_entry)
+
+
+def updaet_benchmark_to_db():
+    '''
+    update daily benchmark weight
+    '''
+    pass
+
+
+async def daily_update():
+    last_update = log.get_time('daily_update')
+    if last_update is None or utils.time_in_beijing() - last_update >= dt.timedelta(days=1):
+        print("running daily update")
+        # check if need to update
+        # update stock price
+        left_fill_stocks_price()
+        right_fill_stock_price()
+        print("updated stocks price")
+        # update benchmark index
+        left_fill_benchmark_profile()
+        right_fill_bechmark_profile()
+        print("updated benchmark profile")
+        # update all stock detail
+        update_stocks_details_to_db()
+        print("updated stocks details")
+        log.update_log('daily_update')
+    else:
+        print("no update needed")
+
+
 def update():
     '''
     run only once, update stock price and benchmark profile
@@ -316,7 +403,7 @@ def update():
     # collect daily stock price until today in beijing time
     if need_to_update_stocks_price(dt.timedelta(days=1)):
         print("Updating stock_price table")
-        stock_df = update_stock_price()
+        # stock_df = update_stock_price()
         stock_df = add_details_to_stock_df(stock_df)
         save_stock_price_to_db(stock_df)
         stock_price_stream.emit(stock_df)

settings.py

@@ -1,6 +1,9 @@
 from datetime import timedelta
 stream_frequency = timedelta(seconds=60)
 
+
+FREQUENCY = timedelta(hours=24)
+
 TABLE_NAME_AND_FREQ = [
     ('benchmark_profile', timedelta(days=1)),
     ('portfolio_profile', timedelta(days=1))