Skip to content

Instantly share code, notes, and snippets.

@AhmedSoliman
Last active January 8, 2023 22:35
Show Gist options
  • Save AhmedSoliman/fb9ce25f39eff227768c0f219e979a3c to your computer and use it in GitHub Desktop.
Save AhmedSoliman/fb9ce25f39eff227768c0f219e979a3c to your computer and use it in GitHub Desktop.
A script to transpose Schwab's transaction CSV to a slightly more friendly format for galloway's https://github.com/mattjgalloway/cgtcalc
import csv
import sys
from typing import Dict, List
ACTION_MAPPING = {
"sell": "SELL",
"buy": "BUY",
"stock plan activity": "BUY",
"reinvest shares": "BUY",
"stock split": "SPLIT",
}
OUT_FIELDS = [
"Date",
"Action",
"Mapped Action",
"Symbol",
"Quantity",
"Price($)",
"Expenses($)",
"Amount($)",
"Price(£)",
"Expenses(£)",
"Amount(£)",
"$->£",
"Description",
]
def main():
if len(sys.argv) < 2:
raise Exception("You must supply the input CSV file")
with open("output.csv", "w", newline="") as out:
writer = csv.DictWriter(out, fieldnames=OUT_FIELDS, dialect='excel')
writer.writeheader()
with open(sys.argv[1]) as f:
header_mapping = {
"Date": 0,
"Action": 1,
"Symbol": 2,
"Description": 3,
"Quantity": 4,
"Price": 5,
"Fees & Comm": 6,
"Amount": 7,
}
for line in csv.reader(f):
assert type(line) is list
if len(line) < 8:
# a header of some sort
continue
if line[0] == "Date":
# Use as header keys.
for index, key in enumerate(line):
if key in header_mapping:
header_mapping[key] = index
continue
if "/" not in line[0]:
# Another header/summary of some sort.
continue
process_line(line, header_mapping, writer)
def process_line(line: List[str], mappings: Dict[str, int], writer: csv.DictWriter):
# Processing lines:
# - Fix the date from MM/DD/YYYY -> DD/MM/YYYY
# - If the date has "as of", use the "as of" date instead, then fix.
# - Map the operations to galloway compatible string in a new column
# - Add empty columns for currency conversions and GBP amount
# - https://github.com/mattjgalloway/cgtcalc
date_pair = line[mappings["Date"]].split(" as of ")
# Convert to DD/MM/YYYY
us_date = date_pair.pop().split("/")
sane_date = f"{us_date[1]}/{us_date[0]}/{us_date[2]}"
action = line[mappings["Action"]]
mapped_action = ACTION_MAPPING.get(action.lower(), "")
writer.writerow(
{
"Date": sane_date,
"Action": action,
"Mapped Action": mapped_action,
"Symbol": line[mappings["Symbol"]],
"Quantity": line[mappings["Quantity"]],
"Price($)": line[mappings["Price"]].replace('$', ''),
"Expenses($)": line[mappings["Fees & Comm"]].replace('$', ''),
"Amount($)": line[mappings["Amount"]].replace('$', ''),
"Price(£)": "TODO",
"Expenses(£)": "TODO",
"Amount(£)": "TODO",
"$->£": "TODO",
"Description": line[mappings["Description"]],
}
)
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment