ledger_prices_db.py

#!/usr/bin/env python
"""Update prices.db with latest data."""

import typing
from datetime import datetime

import pandas as pd

import common
import etfs
import homes
import stock_options

DATE_FORMAT = "%Y/%m/%d %H:%M:%S"
NOW = datetime.now().strftime(DATE_FORMAT)


def main():
    """Main."""
    with common.temporary_file_move(common.LEDGER_PRICES_DB) as output_file:
        # Commodities
        for row in etfs.get_etfs_df().itertuples():
            output_file.write(f"P {NOW} {row.Index} ${row.current_price}\n")

        # Forex values
        forex_df = common.read_sql_table("forex").resample("D").last()
        for row in forex_df.itertuples():
            timestamp = f"{row.Index}".replace("-", "/")
            if pd.notna(row.CHFUSD):
                output_file.write(f"P {timestamp} CHF ${row.CHFUSD}\n")
            if pd.notna(row.SGDUSD):
                output_file.write(f"P {timestamp} SGD ${row.SGDUSD}\n")

        # Properties
        real_estate_df = homes.get_real_estate_df()
        for p in homes.PROPERTIES:
            col = f"{p.name} Price"
            output_file.write(
                f'P {NOW} "{p.address}" ${real_estate_df[col].iloc[-1]:.0f}\n'
            )

        # Stock options
        o: stock_options.OptionsAndSpreads = stock_options.get_options_and_spreads()
        options_df = o.all_options
        commodities_written = set()
        for idx, row in options_df.iterrows():
            idx = typing.cast(tuple, idx)
            name = idx[1]
            if name in commodities_written:
                continue
            value = row["value"] / row["count"]
            output_file.write(f'P {NOW} "{name}" ${value:.2f}\n')
            commodities_written.add(name)

        # Futures
        for row in o.futures.futures_df.reset_index().itertuples():
            if row.commodity in commodities_written:
                continue
            value = row.current_price * row.multiplier  # type: ignore
            output_file.write(f'P {NOW} "{row.commodity}" ${value:.2f}\n')
            commodities_written.add(row.commodity)


if __name__ == "__main__":
    main()