Skip to content

stock_csv.py

CSVDataHandler

Class for handling CSV data.

Source code in pystocktopus/stock_csv.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
class CSVDataHandler:
    """Class for handling CSV data."""

    @staticmethod
    def csv_data_reader(csv_file, csv_stock_column_name: str) -> list[str]:
        """Reads the data from the specified column in a CSV file.

        Args:
            csv_file (str): The path to the CSV file.
            csv_stock_column_name (str): The name of the column to read.

        Returns:
            List[str]: A list of the values in the specified column.
        """
        data_values: list[str] = []

        try:
            logging.info(
                f"Attempting to read column '{csv_stock_column_name}' from CSV file: {csv_file}"
            )
            with open(csv_file) as file:
                csv_reader = csv.reader(file)

                # Read the header row to get column names
                header = next(csv_reader)

                # Find the index of the column
                data_column_index = -1
                for i, col_name in enumerate(header):
                    if (
                        col_name.lower() == csv_stock_column_name.lower()
                    ):  # Case-insensitive comparison
                        data_column_index = i

                # Check if the column was found
                if data_column_index == -1:
                    raise ValueError(f"No {csv_stock_column_name} column found.")

                logging.info(
                    f"Column '{csv_stock_column_name}' found at index {data_column_index}."
                )

                # Read the data from the specified column
                for row in csv_reader:
                    if data_column_index < len(row):
                        data_values.append(row[data_column_index])

                logging.info(
                    f"Successfully read {len(data_values)} rows from column '{csv_stock_column_name}'."
                )

        except FileNotFoundError:
            logging.error(f"File not found: {csv_file}")
            raise FileNotFoundError(f"File not found: {csv_file}")
        except Exception as e:
            logging.error(f"Error reading CSV file: {str(e)}")
            raise e

        return data_values

    @staticmethod
    def _getValue(closing_price: dict[float, float]):
        """Extracts the last value from each key in a dictionary.

        Args:
            closing_price (Dict[float, float]): A dictionary of closing prices.

        Returns:
            List[float]: A list of the last values in the dictionary.
        """
        logging.info("Extracting last values from the closing price dictionary.")
        last_values = []

        for _key, value_list in closing_price.items():
            if value_list:
                last_value = value_list[-1]
                last_values.append(last_value)

        logging.info(
            f"Extracted {len(last_values)} values from closing price dictionary."
        )
        return last_values

    @staticmethod
    def combine_data_csv(
        data_values: list[float], close_list: dict[float, float]
    ) -> list[float]:
        logging.info("Combining data values with closing prices.")
        data_extractor = CSVDataHandler._getValue(close_list)

        results: list[float] = []
        for bought, closing_price in zip(data_values, data_extractor):
            results.append(float(bought) * float(closing_price))

        logging.info(f"Combined data into {len(results)} result values.")
        return results

    @staticmethod
    def update_csv(
        csv_path: str, results: list[float], new_column_name: str = "Price Calculated"
    ) -> None:
        """Updates the CSV file with a new column for the calculated values."""
        try:
            logging.info(
                f"Updating CSV file: {csv_path} with new column '{new_column_name}'."
            )

            df = pd.read_csv(csv_path)
            df[new_column_name] = results

            df.to_csv(csv_path, index=False)

            logging.info(
                f"Successfully added '{new_column_name}' column with {len(results)} values to the CSV file."
            )

        except FileNotFoundError:
            logging.error(f"CSV file not found: {csv_path}")
            raise FileNotFoundError(f"CSV file not found: {csv_path}")
        except Exception as e:
            logging.error(f"Error updating CSV file: {str(e)}")
            raise e

    @staticmethod
    def close_list_csv(
        ticker_data: dict[str, list[float]],
        closing_data_fieldname: list[str] | None = None,
        csv_file_name="stock_data.csv",
    ) -> None:
        """Stores the closing list stock results in a CSV file."""
        if closing_data_fieldname is None:
            closing_data_fieldname = ["closing_stock_data"]

        if not ticker_data:
            logging.warning("Ticker data is empty. No CSV file will be created.")
            return

        try:
            logging.info(f"Writing closing stock data to CSV file: {csv_file_name}")
            with open(csv_file_name, "w", newline="") as csvfile:
                fieldnames = ["Date"] + [
                    f"{field}_{ticker}"
                    for ticker in ticker_data
                    for field in closing_data_fieldname
                ]
                writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
                writer.writeheader()

                num_rows = len(next(iter(ticker_data.values())))
                for i in range(num_rows):
                    data_row = {"Date": f"Date_{i + 1}"}
                    for ticker, close_list in ticker_data.items():
                        for field in closing_data_fieldname:
                            data_row[f"{field}_{ticker}"] = close_list[i]
                    writer.writerow(data_row)

            logging.info(
                f"CSV file '{csv_file_name}' created successfully with {num_rows} rows."
            )
            file_path = os.path.abspath(csv_file_name)
            logging.info(f"CSV file saved at: {file_path}")

        except Exception as e:
            logging.error(f"Error writing closing list data to CSV: {str(e)}")
            raise e

close_list_csv(ticker_data, closing_data_fieldname=None, csv_file_name='stock_data.csv') staticmethod

Stores the closing list stock results in a CSV file.

Source code in pystocktopus/stock_csv.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
@staticmethod
def close_list_csv(
    ticker_data: dict[str, list[float]],
    closing_data_fieldname: list[str] | None = None,
    csv_file_name="stock_data.csv",
) -> None:
    """Stores the closing list stock results in a CSV file."""
    if closing_data_fieldname is None:
        closing_data_fieldname = ["closing_stock_data"]

    if not ticker_data:
        logging.warning("Ticker data is empty. No CSV file will be created.")
        return

    try:
        logging.info(f"Writing closing stock data to CSV file: {csv_file_name}")
        with open(csv_file_name, "w", newline="") as csvfile:
            fieldnames = ["Date"] + [
                f"{field}_{ticker}"
                for ticker in ticker_data
                for field in closing_data_fieldname
            ]
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()

            num_rows = len(next(iter(ticker_data.values())))
            for i in range(num_rows):
                data_row = {"Date": f"Date_{i + 1}"}
                for ticker, close_list in ticker_data.items():
                    for field in closing_data_fieldname:
                        data_row[f"{field}_{ticker}"] = close_list[i]
                writer.writerow(data_row)

        logging.info(
            f"CSV file '{csv_file_name}' created successfully with {num_rows} rows."
        )
        file_path = os.path.abspath(csv_file_name)
        logging.info(f"CSV file saved at: {file_path}")

    except Exception as e:
        logging.error(f"Error writing closing list data to CSV: {str(e)}")
        raise e

csv_data_reader(csv_file, csv_stock_column_name) staticmethod

Reads the data from the specified column in a CSV file.

Parameters:

Name Type Description Default
csv_file str

The path to the CSV file.

 required
csv_stock_column_name str

The name of the column to read.

 required

Returns:

Type Description
list[str]

List[str]: A list of the values in the specified column.
Source code in pystocktopus/stock_csv.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
@staticmethod
def csv_data_reader(csv_file, csv_stock_column_name: str) -> list[str]:
    """Reads the data from the specified column in a CSV file.

    Args:
        csv_file (str): The path to the CSV file.
        csv_stock_column_name (str): The name of the column to read.

    Returns:
        List[str]: A list of the values in the specified column.
    """
    data_values: list[str] = []

    try:
        logging.info(
            f"Attempting to read column '{csv_stock_column_name}' from CSV file: {csv_file}"
        )
        with open(csv_file) as file:
            csv_reader = csv.reader(file)

            # Read the header row to get column names
            header = next(csv_reader)

            # Find the index of the column
            data_column_index = -1
            for i, col_name in enumerate(header):
                if (
                    col_name.lower() == csv_stock_column_name.lower()
                ):  # Case-insensitive comparison
                    data_column_index = i

            # Check if the column was found
            if data_column_index == -1:
                raise ValueError(f"No {csv_stock_column_name} column found.")

            logging.info(
                f"Column '{csv_stock_column_name}' found at index {data_column_index}."
            )

            # Read the data from the specified column
            for row in csv_reader:
                if data_column_index < len(row):
                    data_values.append(row[data_column_index])

            logging.info(
                f"Successfully read {len(data_values)} rows from column '{csv_stock_column_name}'."
            )

    except FileNotFoundError:
        logging.error(f"File not found: {csv_file}")
        raise FileNotFoundError(f"File not found: {csv_file}")
    except Exception as e:
        logging.error(f"Error reading CSV file: {str(e)}")
        raise e

    return data_values

update_csv(csv_path, results, new_column_name='Price Calculated') staticmethod

Updates the CSV file with a new column for the calculated values.

Source code in pystocktopus/stock_csv.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@staticmethod
def update_csv(
    csv_path: str, results: list[float], new_column_name: str = "Price Calculated"
) -> None:
    """Updates the CSV file with a new column for the calculated values."""
    try:
        logging.info(
            f"Updating CSV file: {csv_path} with new column '{new_column_name}'."
        )

        df = pd.read_csv(csv_path)
        df[new_column_name] = results

        df.to_csv(csv_path, index=False)

        logging.info(
            f"Successfully added '{new_column_name}' column with {len(results)} values to the CSV file."
        )

    except FileNotFoundError:
        logging.error(f"CSV file not found: {csv_path}")
        raise FileNotFoundError(f"CSV file not found: {csv_path}")
    except Exception as e:
        logging.error(f"Error updating CSV file: {str(e)}")
        raise e