Coverage for python3/src/rh.py: 100%

1#!../bin/python 

2import robin_stocks.robinhood as r 

3import csv 

4import collections 

5import os 

6from datetime import datetime 

7from dotenv import load_dotenv 

8from typing import List, Optional 

9 

10class CSVParser: 

11 

12 def __init__(self): 

13 pass 

14 

15 def setCSVFile(self, CSVFileName: str): 

16 """ 

17 Set the CSV file name to be used by 

18 the parser. 

19 """ 

20 self.CSVFileName = CSVFileName 

21 

22 def setIgnoreList(self, ignoreList: Optional[List[str]]): 

23 if ignoreList: 

24 self.ignoreList = ignoreList 

25 else: 

26 self.ignoreList = [] 

27 

28 def setProfile(self): 

29 profile = collections.defaultdict(list) 

30 with open('test.csv', 'r') as csvfile: 

31 csvreader = csv.DictReader(csvfile) 

32 for row in reversed(list(csvreader)): 

33 symbol = row["symbol"] 

34 if symbol in self.ignoreList: 

35 continue 

36 side = row['side'] 

37 tranche = {"shares": int(float(row['quantity'])), 

38 "dateAcquired": row['date'], 

39 "avgPrice": float(row['average_price'])} 

40 match side: 

41 case 'buy': 

42 profile[symbol].append(tranche) 

43 case 'sell': 

44 while tranche["shares"] > 0: 

45 profile[symbol][-1]["shares"] -= 1 

46 tranche["shares"] -= 1 

47 if profile[symbol][-1]["shares"] == 0: 

48 profile[symbol].pop() 

49 

50 return profile 

51 

52 

53class Tranche: 

54 

55 def __init__(self, tranche): 

56 self.shares = tranche["shares"] 

57 self.dateAcquired = tranche["dateAcquired"] 

58 self.avgPrice = tranche["avgPrice"] 

59 

60 

61class Stock: 

62 

63 def __init__(self, symbol): 

64 self.symbol = symbol 

65 self.tranches = [] 

66 

67 def createTranche(self, tranche): 

68 """Adds a tranche object to the running list of tranche objects.""" 

69 self.tranches.append(Tranche(tranche)) 

70 

71 def getLongTermCapitalGainsTranche(self): # pragma: no cover 

72 """ 

73 Prints out all stock tranches tranches 

74 that are older the 365 days (one year). 

75 """ 

76 for tranche in self.tranches: 

77 if "." in tranche.dateAcquired: 

78 dateFormat = "%Y-%m-%dT%H:%M:%S.%fZ" 

79 else: 

80 dateFormat = "%Y-%m-%dT%H:%M:%SZ" 

81 dateObject = datetime.strptime(tranche.dateAcquired, dateFormat) 

82 currentDate = datetime.now() 

83 timeDifference = currentDate - dateObject 

84 oneYear = 365 

85 if timeDifference.days >= oneYear: 

86 print(f"\n\tShares: {tranche.shares}\n\t", 

87 f"Date Acquired: {tranche.dateAcquired}\n\t", 

88 f"Avg Price: {tranche.avgPrice}", 

89 sep="", 

90 end="\n") 

91 print("") 

92 

93 def getAllTranches(self): # pragma: no cover 

94 """ 

95 Prints out all tranches in this stock. 

96 """ 

97 for tranche in self.tranches: 

98 print(f"\n\tShares: {tranche.shares}\n\t", 

99 f"Date Acquired: {tranche.dateAcquired}\n\t", 

100 f"Avg Price: {tranche.avgPrice}", 

101 sep="", 

102 end="\n") 

103 

104 

105class Portfolio: 

106 

107 def __init__(self): 

108 self.positions = {} 

109 

110 def createStock(self, symbol: str) -> None: 

111 """ 

112 Creates a stock when given a symbol to identify it by 

113 """ 

114 self.positions[symbol] = Stock(symbol) 

115 

116 def getStock(self, symbol: str) -> Stock: 

117 """ 

118 Get a stock object given a valid symbol 

119 """ 

120 return self.positions[symbol] 

121 

122 def setCSVObj(self, csvFile: str, ignoreList: Optional[List[str]] = None): 

123 self.csvParser = CSVParser() 

124 self.csvParser.setCSVFile(csvFile) 

125 self.csvParser.setIgnoreList(ignoreList) 

126 

127 def getProfile(self): 

128 profile = self.csvParser.setProfile() 

129 

130 for symbol, tranches in profile.items(): 

131 if not tranches: 

132 continue 

133 self.createStock(symbol) 

134 stock = self.getStock(symbol) 

135 for tranche in tranches: 

136 stock.createTranche(tranche) 

137 

138if __name__ == "__main__": # pragma: no cover 

139 load_dotenv() 

140 user = os.getenv('username') 

141 password = os.getenv('password') 

142 mfaCode = input("enter the MFA code") 

143 CSVFILE = 'test.csv' 

144 IGNORELIST = ['GRPN', 'FIT', 'ATVI'] 

145 login = r.login(user, password, mfa_code=mfaCode) 

146 r.export_completed_stock_orders(".", "test.csv") 

147 

148 portfolio = Portfolio() 

149 portfolio.setCSVObj(CSVFILE, IGNORELIST) 

150 portfolio.getProfile() 

151 

152 for symbol, stockObj in portfolio.positions.items(): 

153 print(f'\nSymbol: {symbol}') 

154 stockObj.getAllTranches()