Export Fidelity Positions to www.portfoliovisualizer.com

extend_history.py

pv_export_long_history = '~/Downloads/pv_export_long_history.csv'
overrides = {
    'RSBY': 'LCSIX',
    'LCDS': 'DFSVX',
    'MCDS': 'VOE',
    'SCDS': 'SPYV',
    'NIXT': 'SPY',
    'EZBC': cash,
    'SPYU': 'UPRO',
    'MFUT': 'ASFYX',
    'AHLT': 'ASFYX',
    'TFPN': 'ASFYX',
    'QIS': 'LCSIX',
    'VFLO': 'LEXCX',
    'ASMF': 'ASFYX',
    'CAOS': cash,
    'RSBT': 'ASFYX',
    'HYGI': 'HYG',
    'CTA': 'LCSIX',
    'CRIT': 'XME',
    'DISV': 'DISVX',
    'HGER': 'LCSIX',
    'UPAR': 'DBC',
    'FLSP': 'QSPIX',
    'PFIX': 'TIPZ',
    'KMLM': 'ASFYX',
    'GOVZ': 'ZROZ',
    'CCRV': 'LCSIX',
    'KRBN': cash,
    'FCPI': 'BTAL',
    'AVUV': 'DFSVX',
    'DBMF': 'ASFYX',
    'IVOL': 'TIPZ',
    'DYNF': 'USMV',
    'IETC': 'PPA',
    'DFEN': 'PPA',
    'COWZ': 'LEXCX',
    'NANR': 'IEO',
    'CIBR': cash,
    'AIRR': 'PKB',
    'FMF': 'ASFYX',
    'WTMF': 'ASFYX',
    'USDU': 'UUP',
    'RSBA': 'MNA',
    'AVES': 'DFEVX',
    'GOLY': 'IAU',
}
overrides_df = pl.DataFrame({'Symbol': list(overrides.keys()), 'Symbol Override': list(overrides.values())})
portfolio = (portfolio
    .join(overrides_df, on='Symbol', how='left')
    .group_by(pl.coalesce(pl.col('Symbol Override'), pl.col('Symbol')).alias('Symbol'))
    .agg(pl.sum('Weight'))
    .sort(pl.col('Weight'), descending=True)
)

portfolio.write_csv(pv_export_long_history)
print(f'Wrote {len(portfolio)} securities to {pv_export_long_history}')
portfolio

fidelity_to_pv.py

import polars as pl

fidelity_export = '~/Downloads/Portfolio_Positions_Feb-18-2025.csv'
pv_export = '~/Downloads/pv_export.csv'
account_name = 'Personal'
cash = '^CASHUS'

def money_col(name): return pl.col(name).str.replace_all(r'[$,]', '').cast(pl.Float32)

portfolio = (pl
    .read_csv(fidelity_export)
    .filter(pl.col('Account Name') == account_name)
    .filter(~pl.col('Symbol').str.starts_with(' -')) # remove options
    .with_columns([
        pl.col('Symbol').str.replace_all(r'^(SPAXX|Pending).*', cash),  # Cash positions
        money_col('Last Price').alias('Price'),
        pl.col('Quantity'),
        money_col('Current Value'),
    ])
    # Sometimes current value is empty (e.g when security is in margin)
    .with_columns(pl.coalesce(pl.col('Current Value'), pl.col('Quantity') * pl.col('Price')).alias('Value'))
)

total = portfolio['Value'].sum()
portfolio = (portfolio
    .with_columns((100 * pl.col('Value')/total).alias('Weight'))
    # sometimes the same security is listed multiple times e.g. when it is loaned out
    .group_by(pl.col('Symbol'))
    .agg(pl.sum('Weight'))
    .sort(pl.col('Weight'), descending=True)
)
portfolio.write_csv(pv_export)
print(f'Wrote {len(portfolio)} securities to {pv_export}')
total, portfolio