Coverage for src/sankey_cashflow/transactions.py: 74%

1import datetime 

2from typing import Optional, Union 

3from uuid import uuid4 

4 

5import networkx as nx 

6import pandas as pd 

7from pandas._libs.tslibs import nattype, timestamps 

8 

9from .data_row import DataRow 

10from .utils import df_date_filter, is_empty, logger 

11 

12 

13class Transactions: 

14 """ 

15 Contains transaction data and all helper methods for transforming and outputing. 

16 init with a Pandas dataframe from Google Sheets or a csv. NOTE: depending on the activity, dataframes are mutable. 

17 TODO: 

18 empty values from CSV are 'NaN' 

19 Make sure various synthetic entries don't cause problems for other computations 

20 Make sure that various methods can be run in any order, or enforce precedence/locking 

21 """ 

22 def __init__(self, dataframe, labels_obj, app_settings_obj): 

23 self._df = dataframe 

24 self._grouped_df = None 

25 self.length = len(dataframe) 

26 self._app_settings = app_settings_obj 

27 self._labels_obj = labels_obj 

28 self._validate_df() # Will throw an exception if invalid 

29 # Convert all dates to datetimes and sort earliest to latest 

30 if self._app_settings.verbose: 

31 logger.info(f"Converting data in {self.length} fetched rows to datetimes...") 

32 self._df["Date"] = pd.to_datetime(self._df["Date"]) # Does not mutate dataframe 

33 if nattype.NaTType in [type(i) for i in self._df["Date"]]: 

34 logger.critical(repr(self._df["Date"])) 

35 raise Exception("Empty date found!") # There is probably a better way to do this. 

36 self.earliest_date = self._df["Date"].sort_values().iloc[0] # Returns pandas._libs.tslibs.timestamps.Timestamp 

37 self.latest_date = self._df["Date"].sort_values().iloc[len(self._df) - 1] 

38 self.default_date = self.latest_date - datetime.timedelta(days=1) # Day before latest date in dataset. 

39 self.max_depth = 1 

40 self.tips_processed = False 

41 self.sales_tax_processed = False 

42 self.surplus_deficit_processed = False 

43 self.amount_distributions = False 

44 self.process_report = f"Transactions report\n{'=' * 60}\n\n\n" 

45 

46 @property 

47 def grouped_data(self) -> Union[pd.DataFrame, None]: 

48 """ 

49 The collapsed (Source, Target, Amount) edge list produced by collapse(). None until 

50 process()/process_line() -> collapse() has run. 

51 """ 

52 return self._grouped_df 

53 

54 @property 

55 def processed_data(self) -> pd.DataFrame: 

56 """ 

57 The full per-transaction dataframe, including synthetic rows added during processing. 

58 """ 

59 return self._df 

60 

61 def _validate_df(self) -> bool: 

62 # Validate header row 

63 # WIP: port over new sources-targets column 

64 header_is_valid = DataRow.validate(self._df.columns.to_list(), True) 

65 if not header_is_valid[0]: 

66 raise Exception(f"Source columns failed validation! Error was: {header_is_valid[1]}") 

67 # Validate data rows 

68 amt_types = [isinstance(i, float) for i in self._df["Amount"]] 

69 if False in amt_types: 

70 invalid_loc = amt_types.index(False) # Note, only return first invalid location 

71 raise Exception(f"Invalid data found at row {invalid_loc}!\n {self._df.iloc[invalid_loc]}") 

72 return True 

73 

74 def audit(self, audit_data: pd.DataFrame, date_range: Optional[Union[tuple[str, str], None]] = None) -> str: 

75 """ 

76 Compare transaction data to audit data. Note this is specifically set up to use the column format for my 

77 bank export data. YMMV. 

78 Step 1: Apply date filtering (if applicable) (TODO: Test date handling) 

79 Step 2: Create a column with sums for amount, tax, tips to use for lookups 

80 Step 3: Loop through the bank export and search for matching entries based on the transaction amount 

81 Step 3a: If multiple transactions have the same amount, check that any of them fall +/- 5 days. 

82 Note: this does create a edge case where you could have a false negative if two transaction had the 

83 same values. within the search time. 

84 Write out a report with any suspected missings. Note: this will be a little noisy since I break 

85 transactions into multiple rows sometimes. (eg, Costco visits) 

86 """ 

87 dt_today = datetime.datetime.today() 

88 audit_report = "" 

89 

90 # Step 1 

91 if self._app_settings.filter_dates: 

92 if not date_range: 

93 raise Exception("Filter dates flag was True, but no dates were passed in!") 

94 start_date = date_range[0] 

95 end_date = date_range[1] 

96 if end_date is None and not self._app_settings.all_time: 

97 end_date = dt_today 

98 if start_date is None and not self._app_settings.all_time: 

99 start_date = self._app_settings.DEFAULT_START_DATE 

100 audit_data = df_date_filter(audit_data, start_date, end_date) 

101 elif not self._app_settings.all_time: 

102 audit_data = df_date_filter(audit_data, self._app_settings.DEFAULT_START_DATE, dt_today) 

103 

104 def safe_sum(pd_series): 

105 """ 

106 Safely sum a transaction frame 

107 """ 

108 def vval(val): 

109 if not val: 

110 return 0 

111 return round(float(val), 2) 

112 return round(pd_series["Amount"] + vval(pd_series.get("Sales Tax")) + vval(pd_series.get("Tips")), 2) 

113 

114 # Step 2 

115 rowsums = self._df.apply(safe_sum, axis=1) 

116 

117 # Step 3 

118 for idx, row in audit_data.iterrows(): 

119 transaction_found = False 

120 hits = self._df[rowsums == row["Amount"]] 

121 for hit in hits["Date"]: 

122 if (row["Date"] < (hit + datetime.timedelta(days=5))) and \ 

123 (row["Date"] > (hit - datetime.timedelta(days=5))): 

124 transaction_found = True 

125 if not transaction_found: 

126 audit_report += f"Transaction not found for {row['Date']} - {row['Description']} - {row['Amount']}\n" 

127 

128 return audit_report 

129 

130 def process(self, date_range=None): 

131 """ 

132 Process dataframe for sankey diagram 

133 Step 1: Drop rows based on tag exclusions 

134 Step 2: Split out any entries containing distributions (if feature flag is turned on) 

135 Step 3: Apply date filtering (if applicable) 

136 Step 4: Update transaction rows with sources and targets as defined by labels spreadsheet for all entries, 

137 modify sources/targets based on tags & store filters. Also handle recurring items. 

138 Step 5: Loop through each entry and: 

139 a: check for sales tax and/or tips column. Update total amount and create synthetic flows for tax/tips. 

140 b: crawl back through DAG, creating synthetic entries for predecessor nodes along the way, 

141 to ensure flows appear correctly. 

142 Step 6: Compute surplus or deficit flows 

143 Step 7: aggregate amounts for all shared source:target pairs 

144 """ 

145 

146 dt_today = datetime.datetime.today() 

147 self.process_report += f"Processing {len(self._df)} transactions \ 

148 from {self._app_settings.source_data_location()}\n{'-' * 60}\n" 

149 if self._app_settings.verbose: 

150 logger.info(f"Processing {len(self._df)} transactions from {self._app_settings.source_data_location()}") 

151 

152 # Step 1: 

153 if self._app_settings.exclude_tags: 

154 self.filter_tags(self._app_settings.exclude_tags) 

155 

156 # Step 2: 

157 if self._app_settings.distribute_amounts: 

158 if self._app_settings.verbose: 

159 logger.info("Distributing amounts...") 

160 self.distribute_amounts() # process report logging happens in called method 

161 

162 # Step 3: 

163 if self._app_settings.filter_dates: 

164 if not date_range: 

165 raise Exception("Filter dates flag was True, but no dates were passed in!") 

166 start_date = date_range[0] 

167 end_date = date_range[1] 

168 if end_date is None and not self._app_settings.all_time: 

169 end_date = dt_today 

170 if start_date is None and not self._app_settings.all_time: 

171 start_date = self._app_settings.DEFAULT_START_DATE 

172 self.process_report += f"Filtering for dates from {start_date} to {end_date}\n{'-' * 60}\n" 

173 if self._app_settings.verbose: 

174 logger.info(f"Filtering for dates from {start_date} to {end_date}") 

175 # TODO: test and find edge cases! 

176 self.filter_dates(start_date, end_date) 

177 elif not self._app_settings.all_time: 

178 self.filter_dates(self._app_settings.DEFAULT_START_DATE, dt_today) 

179 

180 self.update_title() 

181 # Step 4: 

182 self.apply_labels() 

183 # Step 5: 

184 self.process_rows() 

185 # Step 6: 

186 self.create_surplus_deficit_flows() 

187 # Step 7: 

188 self.collapse() 

189 # -- END Transactions.process() -- 

190 

191 def process_line(self, date_range=None): 

192 """ 

193 Process dataframe for line chart diagram 

194 Step 1: Drop rows based on tag exclusions 

195 Step 2: Split out any entries containing distributions (if feature flag is turned on) 

196 Step 3: Apply date filtering (if applicable) 

197 Step 4: Update transaction rows with sources and targets as defined by labels spreadsheet for all entries, 

198 modify sources/targets based on tags & store filters. Also handle recurring items. 

199 """ 

200 

201 dt_today = datetime.datetime.today() 

202 self.process_report += f"Processing {len(self._df)} transactions from \ 

203 {self._app_settings.source_data_location()}\n{'-' * 60}\n" 

204 if self._app_settings.verbose: 

205 logger.info(f"Processing {len(self._df)} transactions from {self._app_settings.source_data_location()}") 

206 

207 # Step 1: 

208 if self._app_settings.exclude_tags: 

209 self.filter_tags(self._app_settings.exclude_tags) 

210 

211 # Step 2: 

212 if self._app_settings.distribute_amounts: 

213 if self._app_settings.verbose: 

214 logger.info("Distributing amounts...") 

215 self.distribute_amounts() # process report logging happens in called method 

216 

217 # Step 3: 

218 if self._app_settings.filter_dates: 

219 if not date_range: 

220 raise Exception("Filter dates flag was True, but no dates were passed in!") 

221 start_date = date_range[0] 

222 end_date = date_range[1] 

223 if end_date is None and not self._app_settings.all_time: 

224 end_date = dt_today 

225 if start_date is None and not self._app_settings.all_time: 

226 start_date = self._app_settings.DEFAULT_START_DATE 

227 self.process_report += f"Filtering for dates from {start_date} to {end_date}\n{'-' * 60}\n" 

228 if self._app_settings.verbose: 

229 logger.info(f"Filtering for dates from {start_date} to {end_date}") 

230 # TODO: test and find edge cases! 

231 self.filter_dates(start_date, end_date) 

232 elif not self._app_settings.all_time: 

233 self.filter_dates(self._app_settings.DEFAULT_START_DATE, dt_today) 

234 

235 # Step 4: 

236 self.apply_labels() 

237 # -- END Transactions.process_line() -- 

238 

239 def filter_tags(self, tags_to_exclude): 

240 # tags_to_exclude: ['tag1','tag2', ...] 

241 self.process_report += f"Checking for tags to exclude: {tags_to_exclude}\n{'-' * 60}\n" 

242 df_changed = False 

243 rows_to_drop = [] 

244 for k, v in enumerate(self._df["Tags"]): 

245 tag_matches = DataRow.tag_matches(v, tags_to_exclude) # None if either arg is None or if no matches 

246 if tag_matches: 

247 df_changed = True 

248 rows_to_drop.append(self._df.index[k]) 

249 if df_changed and rows_to_drop: # Do as a separate loop to avoid changing the frame as we're iterating over it. 

250 for row_idx in rows_to_drop: 

251 self.process_report += f"DROPPING row due to exclude tags: {self._df.loc[row_idx]}\n" 

252 if self._app_settings.verbose: 

253 logger.info(f"DROPPING row due to exclude tags: {self._df.loc[row_idx]}") 

254 self._df.drop(row_idx, inplace=True) 

255 if df_changed: 

256 self._df.reset_index(inplace=True, drop=True) 

257 

258 def add_row(self, row_data, already_validated=False): 

259 try: 

260 idx = len(self._df) # Could use self.length... 

261 if already_validated: 

262 self._df.loc[idx] = row_data 

263 else: 

264 self._df.loc[idx] = DataRow.validate(row_data) 

265 self.length = idx + 1 

266 except Exception as e: 

267 logger.error(f"Error adding row: {row_data} - {e}") 

268 raise 

269 

270 def apply_labels(self): 

271 """ 

272 Loop through each row in dataframe, looking up source-target nodes using category names, and overriding if 

273 indicated by tags or stores flags. 

274 Also add each source:target pair as an edge in a DAG, to be used in the sankey diagram generator to create 

275 intermediate transactions. 

276 NOTE: this process will not be adding intermediate transactions, but will ensure the DAG is correct so that 

277 intermediate transactions can be added later. 

278 """ 

279 self.process_report += f"Running Transactions.apply_labels(). Tags has: {self._app_settings.tags}, \ 

280 tag_override is {self._app_settings.tag_override} and stores has: {self._app_settings.stores}\n\n" 

281 if self._app_settings.tags and self._app_settings.verbose: 

282 logger.info(f"Tag search enabled: Looking for tags: {self._app_settings.tags}") 

283 if self._app_settings.tag_override: 

284 logger.info("Overriding tags...") 

285 if self._app_settings.stores and self._app_settings.verbose: 

286 logger.info(f"Store search enabled: Looking for stores: {self._app_settings.stores}") 

287 if self._app_settings.verbose: 

288 logger.info(f"Applying labels for {len(self._df)} transactions") 

289 

290 if self._app_settings.recurring: 

291 if self._app_settings.verbose: 

292 logger.info("Recurring transactions to be split out") # TODO: precludes tag:recurring handling. 

293 # Add edge from Income to Recurring 

294 self._labels_obj._digraph.add_edge("Income", "Recurring") 

295 

296 # Util functions ................................................................................... 

297 def get_source_target_labels(this_obj, this_category_key, this_category_val, step_id): 

298 # Get default labels defined for category from sources-targets sheet, override from data sheet if set there. 

299 src = this_obj._labels_obj.get_attribute(this_category_val, "source") 

300 tgt = this_obj._labels_obj.get_attribute(this_category_val, "target") 

301 classification = this_obj._labels_obj.get_attribute(this_category_val, "classification") 

302 this_obj.process_report += f"[{step_id}] Found default src:target for {this_category_val} -> {src}:{tgt} \ 

303 (classification: {classification})\n" 

304 # Allow individual transaction rows to override label lookups 

305 data_override_s_t = False 

306 transaction_source = this_obj._df.at[this_category_key, "Source"] 

307 transaction_target = this_obj._df.at[this_category_key, "Target"] 

308 if not is_empty(transaction_source) and not is_empty(transaction_target): 

309 # Both a source and target were specifed in the transaction data 

310 src = transaction_source 

311 tgt = transaction_target 

312 data_override_s_t = True 

313 elif not is_empty(transaction_source) and is_empty(transaction_target): 

314 # A source but not target were specifed in the transaction data 

315 src = transaction_source 

316 data_override_s_t = True 

317 elif is_empty(transaction_source) and not is_empty(transaction_target): 

318 # A Target but not source were specifed in the transaction data, 

319 # we will append it to the default source-target 

320 if transaction_target != tgt: # Skip if the override is the same as the default target 

321 if not this_obj._labels_obj._digraph.has_edge(src, tgt): 

322 this_obj._labels_obj._digraph.add_edge(src, tgt) 

323 src = tgt 

324 tgt = transaction_target 

325 data_override_s_t = True 

326 

327 if data_override_s_t: 

328 this_obj.process_report += f"[{this_step_id}] Override source/target for {this_category_val} from \ 

329 transaction data -> {src}: {tgt}\n" 

330 

331 return src, tgt, classification 

332 

333 # MAIN LOOP ........................................................................................ 

334 

335 for k, v in enumerate(self._df["Category"]): 

336 # Main labeling loop. Iterate over each transaction, look up source-target information in labels 

337 # spreadsheet applying/overriding as indicated. 

338 this_step_id = str(uuid4())[:8] 

339 self.process_report += f"[{this_step_id}] START Processing {self._df.at[k, 'Date']} | \ 

340 {v} | {self._df.at[k, 'Tags']} | ${self._df.at[k, 'Amount']}\n" 

341 is_deduction = False 

342 if is_empty(v): 

343 # Note: this should not happen... raise exception instead? 

344 self.process_report += f"[{this_step_id}] SKIPPING empty category {self._df.loc[k]}\n" 

345 logger.info(f"Skipping empty category {self._df.loc[k]}") 

346 continue 

347 # Tag logic 

348 # TODO: test source tags 

349 

350 this_source, this_target, this_classification = get_source_target_labels(self, k, v, this_step_id) 

351 

352 # Handle deduction types (these go directly from a income to an expense, skipping the 'Income' category 

353 # and have a variable source based on their description) 

354 # Use case is a transaction with income would normally be something like "My Job" -> "Income" and then a 

355 # second transaction with income taxes would be "My Job" -> "Income Taxes" (skipping income category) 

356 # TODO: Verify this works correctly with s-tags 

357 if this_source == "DEDUCTIONS": 

358 this_source = self._df.at[k, "Description"] 

359 self._df.at[k, "Type"] = "deduction" 

360 is_deduction = True 

361 self.process_report += f"[{this_step_id}] Deduction type found src:target -> \ 

362 {this_source}:{this_target}\n" 

363 if self._app_settings.verbose: 

364 logger.info(f"Found DEDUCTION type transaction. Set to {this_source}:{this_target}") 

365 

366 if self._app_settings.recurring and this_source == "Income": 

367 # Replace with recurring 

368 this_source = "Recurring" 

369 # Verify base edge is in DAG 

370 if not self._labels_obj._digraph.has_edge(this_source, this_target): 

371 logger.info(f"Edge: {this_source}:{this_target} not found! Adding to graph.") 

372 self._labels_obj._digraph.add_edge(this_source, this_target) 

373 

374 # For now logic around tags/stores with deductions flow is undefined - skip processing. 

375 if not is_deduction: 

376 # Check for store match 

377 store_matches = False 

378 if self._app_settings.stores: 

379 this_name = self._df.at[k, "Description"] 

380 if self._app_settings.stores and this_name in self._app_settings.stores: 

381 store_matches = True 

382 

383 # Check for tag match(es) 

384 # Note currently we will only ever use the first match. 

385 tag_matches = DataRow.tag_matches(self._df.at[k, "Tags"], self._app_settings.tags) 

386 # None if flag is not enabled or no matches 

387 tag_type = None 

388 if tag_matches: 

389 if self._app_settings.recurring and tag_matches and tag_matches[0] == "Recurring": 

390 raise Exception("Not double processing recurring tags!") # TODO: handle more quietly 

391 if self._app_settings.verbose: 

392 logger.info(f"Got tag matches: {tag_matches}") 

393 # Check for s-tags 

394 # Get tag type 

395 tag_type = self._labels_obj.get_attribute(tag_matches[0], "type") 

396 if tag_type == "s-tag": 

397 # If the flow is directly to/from "Income", replace "Income" with the Tag 

398 if this_source == "Income": 

399 this_source = tag_matches[0] 

400 elif this_target == "Income": 

401 this_target = tag_matches[0] 

402 else: 

403 self._labels_obj._digraph.add_edge(this_source, this_target, type="s-tag") 

404 # NOTE: if tag is distant from "Income", we'll need to handle it while reconciling DAG 

405 elif self._app_settings.tag_override: 

406 # If overriding tags, we'll use the labels sheet to determine placement 

407 this_source = self._labels_obj.get_attribute(tag_matches[0], "source", labeltype="tag", 

408 use_default=False) 

409 this_target = self._labels_obj.get_attribute(tag_matches[0], "target", labeltype="tag", 

410 use_default=False) 

411 if not this_source: 

412 # If we don't have matching tag defined in sources-targets sheet, 

413 # just create it to/from income 

414 # lookup the target we would have without tag matching 

415 def_target = self._labels_obj.get_attribute(v, "target", use_default=False) 

416 if def_target == "Income": # Note: Breaks if we have income flows more than one deep 

417 this_source = tag_matches[0] 

418 this_target = "Income" 

419 else: 

420 this_source = "Income" 

421 this_target = tag_matches[0] 

422 else: 

423 # We are appending the tags as new target to the end of the flow 

424 this_source = this_target 

425 this_target = tag_matches[0] 

426 if self._app_settings.verbose: 

427 logger.info(f"Adding edge to graph for tag ({tag_matches}[0]): \ 

428 {this_source} -> {this_target}") 

429 # self._labels_obj._digraph.add_edge(this_source, this_target) 

430 

431 if store_matches: 

432 this_source = this_target 

433 this_target = self._df.at[k, "Description"] 

434 if self._app_settings.verbose: 

435 logger.info(f"Adding edge to graph for store ({this_name}): {this_source} -> {this_target}") 

436 # self._labels_obj._digraph.add_edge(this_source, this_target) 

437 

438 self.process_report += f"[{this_step_id}] RESOLVED src:target for {v} -> {this_source}:{this_target}\n" 

439 if self._app_settings.verbose: 

440 logger.info(f"RESOLVED source/target > {this_source}:{this_target}") 

441 # Circuit breaker 

442 if is_empty(this_source) or is_empty(this_target): 

443 raise Exception(f"Got empty source or target for category {v}! ({this_source}:{this_target})") 

444 

445 # Check for final edge in DAG and add if necessary 

446 if not self._labels_obj._digraph.has_edge(this_source, this_target): 

447 logger.info(f"Edge: {this_source}:{this_target} not found! Adding to graph.") 

448 self._labels_obj._digraph.add_edge(this_source, this_target) 

449 

450 # Sanity check that we haven't created an orphan edge 

451 if not (is_deduction or tag_type == 's-tag') and \ 

452 "Income" not in nx.ancestors(self._labels_obj._digraph, this_target) and \ 

453 "Income" not in nx.descendants(self._labels_obj._digraph, this_source): 

454 logger.debug(f"{self._df.loc[k]}") 

455 raise Exception(f"No path to \'Income\' from {this_source}:{this_target}") 

456 

457 # Set source-target + classification on original transaction 

458 self._df.at[k, "Source"] = this_source 

459 self._df.at[k, "Target"] = this_target 

460 self._df.at[k, "Classification"] = this_classification 

461 

462 self.process_report += f"[{this_step_id}] FINISHED processing labels\n" 

463 

464 def process_rows(self): 

465 """ 

466 Process individual transactions, creating synthetic transactions as needed to satisfy flows 

467 """ 

468 msg = f"Processing row data on {len(self._df)} rows" 

469 self.process_report += f"\n{'-' * 60}\nRunning Transactions.process_rows()\n{'-' * 60}\n" 

470 self.process_report += msg + "\n" 

471 if self._app_settings.verbose: 

472 logger.info(msg) 

473 for k, v in enumerate(self._df["Source"]): 

474 this_row = self._df.loc[k] 

475 this_step_id = str(uuid4())[:8] 

476 is_recurring = False 

477 self.process_report += f"[{this_step_id}] START Processing {self._df.at[k, 'Date']} | \ 

478 {self._df.at[k, 'Description']} | ${self._df.at[k, 'Amount']}\n" 

479 if self._app_settings.verbose: 

480 logger.info(f"{'-' * 40}\nGot a transaction: {self._df.at[k, 'Date']} | \ 

481 {self._df.at[k, 'Description']} | {self._df.at[k, 'Source']}:{self._df.at[k, 'Target']} | \ 

482 ${self._df.at[k, 'Amount']}\n") 

483 

484 # Check for tag match(es) 

485 # Note currently we will only ever use the first match. 

486 # TODO: discard tag if tag is "Recurring" AND _app_settings.recurring is True 

487 # None if flag is not enabled or no matches 

488 tag_matches = DataRow.tag_matches(self._df.at[k, "Tags"], self._app_settings.tags) 

489 tag_type = None 

490 if tag_matches: 

491 tag_type = self._labels_obj.get_attribute(tag_matches[0], "type") # Get tag type 

492 

493 # Check for recurring tag 

494 has_recurring = DataRow.tag_matches(self._df.at[k, "Tags"], ["Recurring"]) 

495 if self._app_settings.recurring and has_recurring and "Recurring" in has_recurring: 

496 is_recurring = True 

497 if self._app_settings.verbose: 

498 logger.info(">> Processing recurring transaction") 

499 

500 # Handle taxes 

501 if not is_empty(this_row["Sales Tax"], True): 

502 if self._app_settings.separate_taxes: 

503 # Add sales tax to it's own root category as a new row 

504 self.process_report += f"[{this_step_id}] ADDED: {this_row.Date} | {this_row.Description} | \ 

505 'Income' -> 'Sales Tax' | ${this_row['Sales Tax']}\n" 

506 self.add_row(DataRow.create( 

507 date=this_row.Date, 

508 category_name="Sales Tax", 

509 amount=this_row["Sales Tax"], 

510 source="Income", 

511 target="Sales Tax", 

512 description=this_row.Description, 

513 tags=this_row.Tags, 

514 comment='Synthetic row for sales tax', 

515 distribution=this_row.Distribution, 

516 classification=self._app_settings.sales_tax_classification 

517 ), True) 

518 else: 

519 # Create new sales tax child target from this original target row & 

520 # add sales tax back to original row amount 

521 # Note: if store or tag processing is being done, this may already be one removed from 

522 # the original category 

523 self.process_report += f"[{this_step_id}] ADDED: {this_row.Date} | {this_row.Description} | \ 

524 {this_row.Target} -> 'Sales Tax' | ${this_row['Sales Tax']}\n" 

525 if not is_empty(this_row["Sales Tax"], True): 

526 self.add_row(DataRow.create( 

527 date=this_row.Date, 

528 category_name="Sales Tax", 

529 amount=this_row["Sales Tax"], 

530 source=this_row.Target, 

531 target="Sales Tax", 

532 description=this_row.Description, 

533 tags=this_row.Tags, 

534 comment='Synthetic row for sales tax', 

535 distribution=this_row.Distribution, 

536 classification=self._app_settings.sales_tax_classification 

537 ), True) 

538 # For this to behave as expected, it needs to add the sales tax amount back 

539 # to the original Amount 

540 self._df.at[k, "Amount"] = round(this_row.Amount + this_row["Sales Tax"], 2) 

541 self.process_report += f"[{this_step_id}] UPDATED: {this_row.Date} | {this_row.Description} | \ 

542 {this_row.Source} -> {this_row.Target} | \ 

543 ${this_row.Amount} -> ${self._df.at[k, 'Amount']}\n" 

544 

545 # Handle tips by creating new Tips child target from this original target row & add tip back 

546 # to original row amount 

547 # Note: if store or tag processing is being done, this may already be one removed from the original category 

548 if not is_empty(this_row["Tips"], True): 

549 self.process_report += f"[{this_step_id}] ADDED: {this_row.Date} | {this_row.Description} | \ 

550 {this_row.Target} -> 'Tips' | ${this_row['Tips']}\n" 

551 # Sales tax computation may have changed from this_row.Amount value 

552 orig_amount = self._df.at[k, "Amount"] 

553 self.add_row(DataRow.create( 

554 date=this_row.Date, 

555 category_name="Tips", 

556 amount=this_row["Tips"], 

557 source=this_row.Target, 

558 target="Tips", 

559 description=this_row.Description, 

560 tags=this_row.Tags, 

561 comment='Synthetic row for tips', 

562 distribution=this_row.Distribution, 

563 classification=self._app_settings.tip_classification 

564 ), True) 

565 # For this to behave as expected, it needs to add the tips amount back to the original Amount 

566 self._df.at[k, "Amount"] = round(orig_amount + this_row["Tips"], 2) 

567 self.process_report += f"[{this_step_id}] UPDATED: {this_row.Date} | {this_row.Description} | \ 

568 {this_row.Source} -> {this_row.Target} | ${orig_amount} -> ${self._df.at[k, 'Amount']}\n" 

569 

570 # Traverse DAG from row source back to Income, adding a synthetic row for each edge it finds. 

571 # NOTE: if using s-tags, will go back to the tag instead of Income 

572 # TODO: explore cases where we are multiple synthetic rows deep, or a synthetic row has been added that 

573 # flows INTO income, or orphan flows (eg deductions) include synthetic nodes. 

574 s_tag_d1 = False 

575 if self._df.at[k, "Type"] == 'deduction': # deduction types skip DAG processing for now 

576 self.process_report += f"[{this_step_id}] SKIPPING DAG traversal, since this is a deduction type.\n" 

577 if self._app_settings.verbose: 

578 logger.info(f"Skipping DAG checks as this was a deductions type entry: {this_row.Date} | \ 

579 {this_row.Description} | {this_row.Source} -> {this_row.Target} | ${this_row.Amount}") 

580 continue 

581 

582 # traverse graph 

583 if tag_type == 's-tag' and (this_row.Source == tag_matches[0] or this_row.Target == tag_matches[0]): 

584 # First order edge and is s-tag - skip processing 

585 s_tag_d1 = True 

586 elif "Income" in nx.ancestors(self._labels_obj._digraph, this_row.Target): 

587 # Must be an expense category: 

588 start_node = "Income" 

589 # This breaks if there are multiple paths to the end node, eg when using tags/stores flows 

590 end_node = this_row.Source 

591 else: 

592 # Must be an income category 

593 start_node = this_row.Source 

594 end_node = "Income" 

595 

596 if not s_tag_d1: 

597 self.process_report += f"[{this_step_id}] Starting to traverse DAG for {start_node} -> {end_node}\n" 

598 if self._app_settings.verbose: 

599 logger.info(f"Traversing graph for {start_node}:{end_node}...") 

600 

601 pgroups = [i for i in nx.all_simple_edge_paths(self._labels_obj._digraph, start_node, end_node)] 

602 if is_recurring: 

603 new_groups = [[]] 

604 for g in pgroups[0]: 

605 if g[0] == "Income": 

606 if self._app_settings.verbose: 

607 logger.info(f"--- Injecting Income:Recurring and Recurring:{g[1]} nodes ----") 

608 new_groups[0].append(("Income", "Recurring")) 

609 new_groups[0].append(("Recurring", g[1])) 

610 else: 

611 new_groups[0].append(g) 

612 pgroups = new_groups 

613 

614 self.process_report += f"[{this_step_id}] DAG search yielded groups: {pgroups}\n" 

615 if self._app_settings.verbose: 

616 logger.info(f"Searched DAG for {start_node} -> {end_node} and got group: {pgroups}...") 

617 if len(pgroups) != 1: 

618 # Potentially an error condition. Maybe raise an exception 

619 logger.info(f"Edge paths search did not yield the expected number of groups! {pgroups}") 

620 for pgroup in pgroups: 

621 # Each edge path will be an array of tuples, like [(source1,target1), (source2,target2), ...] 

622 # Iterate over the paths (ignoring the one that matches the original entry) and create synthetic 

623 # entries for each one. 

624 for pitem in pgroup: 

625 # Don't need to process the pair we already have 

626 if pitem == (this_row.Source, this_row.Target): 

627 continue 

628 syn_source, syn_target = pitem 

629 if syn_source == "Income" and tag_type == 's-tag': 

630 # Since this is an s-tag flow, the root of the flow should be the tag 

631 syn_source = tag_matches[0] 

632 if syn_target == "Income" and tag_type == 's-tag': 

633 syn_target = tag_matches[0] # TODO: verify that this case is handled as expected 

634 self.process_report += f"[{this_step_id}] ADDED: {this_row.Date} | {this_row.Description} | \ 

635 {syn_source} -> {syn_target} | ${self._df.at[k, 'Amount']}\n" 

636 if self._app_settings.verbose: 

637 logger.info(f"Adding synthetic entry: {this_row.Date} | {this_row.Description} | \ 

638 {syn_source} -> {syn_target} | ${self._df.at[k, 'Amount']}") 

639 self.add_row(DataRow.create( 

640 date=this_row.Date, 

641 category_name=this_row.Category, 

642 amount=self._df.at[k, "Amount"], 

643 source=syn_source, 

644 target=syn_target, 

645 description=this_row.Description, 

646 tags=this_row.Tags, 

647 comment='Synthetic row', 

648 distribution=this_row.Distribution, 

649 classification="Uncategorized" 

650 ), True) 

651 

652 self.process_report += f"[{this_step_id}] DONE processing.\n{'-' * 40}\n" 

653 

654 def collapse(self): 

655 self.process_report += f"\n{'-' * 40}\nStepping into Transactions.collapse()\n{'-' * 40}\n" 

656 if self._app_settings.verbose: 

657 logger.info("Aggregating all source-target pairs") 

658 # Collapse all the pairs down for cleaner flows 

659 grouped_df = self._df.groupby(['Source', 'Target']).agg({'Amount': 'sum'}) 

660 # Resetting an index appears to just create a new one unless the drop argument is passed in, 

661 # but that's fine in this case. 

662 grouped_df.reset_index(inplace=True) 

663 self._grouped_df = grouped_df # TODO: Review grouped_df vs _df 

664 if self._app_settings.verbose: 

665 logger.info(f"Collapsed {len(self._df)} transactions down to {len(self._grouped_df)}") 

666 

667 def create_surplus_deficit_flows(self): 

668 self.process_report += f"\n{'-' * 40}\nStepping into Transactions.create_surplus_deficit_flows()\n{'-' * 40}\n" 

669 if self.surplus_deficit_processed: 

670 logger.info("Surplus/deficit flows have already been processed!") 

671 return 

672 self.surplus_deficit_processed = True 

673 if self._app_settings.verbose: 

674 logger.info("Computing source/deficit flows") 

675 # Check for s-tag nodes 

676 # Returns a dict like: {'a': 's-tag', 'd': 's-tag, 'c': 'tag', ...} 

677 node_types = nx.get_node_attributes(self._labels_obj._digraph, 'type') 

678 s_nodes = [i for i in node_types if node_types[i] == 's-tag'] # A list of s-nodes 

679 s_nodes.append("Income")