Skip to content

Analysis

card_stats(database_path, cid)

Calculates the market value and gain/loss of the input card using the timestamps the prices were recorded in the local SQL database.

Parameters:

Name Type Description Default
database_path Path

Path to the local SQL collection database.

 required
cid int

cid of card in collection file.

 required

Returns:

Type Description
DataFrame

Returns dataframe with schema {"timestamp": pl.String, "market_value": pl.Float64, "profit": pl.Float64}.
Source code in src/mtg_pynance/analysis.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
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
def card_stats(database_path: Path, cid: int) -> pl.DataFrame:
    """
    Calculates the market value and gain/loss of the input card using the
    timestamps the prices were recorded in the local SQL database.

    Parameters
    ----------
    database_path: Path
        Path to the local SQL collection database.
    cid: int
        cid of card in collection file.

    Returns
    -------
    pl.DataFrame
        Returns dataframe with schema {"timestamp": pl.String,
        "market_value": pl.Float64, "profit": pl.Float64}.
    """
    # TODO this function assumes that if a card has no price info it has
    # no it has no price table. This is true within the package framework,
    # but may not be if a user pokes around.

    # Connect to local SQL database
    connection: sqlite3.Connection = sqlite3.connect(database_path)
    cursor: sqlite3.Cursor = connection.cursor()

    # Get card's purchase price
    sql_command = f"select price from purchase_price where cid = {cid}"
    purchase_price: Optional[tuple[int, float]] = cursor.execute(sql_command).fetchone()
    if purchase_price is None:
        return f"No records of collection ID {cid} exist in database!"
    purchase_price = float(purchase_price[0])

    # Get card's price data
    sql_command = f"select * from card_{cid}"
    price_data: list[tuple[str, float]] = cursor.execute(sql_command).fetchall()
    card_stats = [
        [timestamp, value, value - purchase_price] for timestamp, value in price_data
    ]

    # Convert stats from list of lists to dataframe
    df = pl.DataFrame(
        card_stats,
        schema={
            "timestamp": pl.String,
            "market_value": pl.Float64,
            "gain/loss": pl.Float64,
        },
        orient="row",
    )

    return df

collection_extrema(database_path)

Calculates the cards in the collection that currently have the largest gain and loss.

Parameters:

Name Type Description Default
database_path Path

Path to the local SQL collection database.

 required

Returns:

Type Description
tuple[dict, dict]

Return a tuple of dictionaries where element 0 corresponds to the card with the maximum gain and element 1 corresponds to the card with the maximum loss. The schema for each dictionary is {"cid": [int], "*": float, "purchase_price": float} where * is "gain" or "loss", respectively.
Source code in src/mtg_pynance/analysis.py 
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
def collection_extrema(database_path: Path) -> tuple[dict, dict]:
    """
    Calculates the cards in the collection that currently have the largest gain
    and loss.

    Parameters
    ----------
    database_path: Path
        Path to the local SQL collection database.

    Returns
    -------
    tuple[dict, dict]
        Return a tuple of dictionaries where element 0 corresponds to the card
        with the maximum gain and element 1 corresponds to the card with the
        maximum loss. The schema for each dictionary is {"cid": [int], "*": float,
        "purchase_price": float} where * is "gain" or "loss", respectively.

    """
    # Connect to local SQL database
    connection: sqlite3.Connection = sqlite3.connect(database_path)
    cursor: sqlite3.Cursor = connection.cursor()

    # Get list of all cards
    sql_command = "select * from purchase_price"
    card_ids: list[tuple[int, float]] = cursor.execute(sql_command).fetchall()

    gain = {"cid": [], "gain": 0.0, "purchase_price": 0.0}
    loss = {"cid": [], "loss": 0.0, "purchase_price": 0.0}

    for _, (cid, purchase_price) in enumerate(card_ids):
        # Calculate gain/loss of each card with respect to current price
        sql_command = f"select market_value from card_{cid} where timestamp = (select max(timestamp) from card_{cid})"
        current_price = float(cursor.execute(sql_command).fetchone()[0])
        gain_loss = round(current_price - purchase_price, 2)

        # Record card's gain/loss if it is an extreme
        if gain_loss == gain["gain"]:
            gain["cid"].append(cid)

        if gain_loss > gain["gain"]:
            gain["cid"].clear()
            gain["cid"].append(cid)
            gain["gain"] = gain_loss
            gain["purchase_price"] = purchase_price

        if gain_loss == loss["loss"]:
            loss["cid"].append(cid)

        if gain_loss < loss["loss"]:
            loss["cid"].clear()
            loss["cid"].append(cid)
            loss["loss"] = gain_loss
            loss["purchase_price"] = purchase_price

    connection.close()

    return gain, loss

collection_largest_movers(database_path)

Calculates the price movement of each card based on its two most recently recorded prices and returns the cards that have the largest and smallest price movement.

Parameters:

Name Type Description Default
database_path Path

Path to the local SQL collection database.

 required

Returns:

Type Description
tuple[dict, dict]

Return a tuple of dictionaries where element 0 corresponds to the card with the maximum gain and element 1 corresponds to the card with the maximum loss. The schema for each dictionary is {"cid": [int], "*": float, "purchase_price": float} where * is "gain" or "loss", respectively.
Source code in src/mtg_pynance/analysis.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
def collection_largest_movers(database_path: Path) -> tuple[dict, dict]:
    """
    Calculates the price movement of each card based on its two most recently recorded
    prices and returns the cards that have the largest and smallest price movement.

    Parameters
    ----------
    database_path: Path
        Path to the local SQL collection database.

    Returns
    -------
    tuple[dict, dict]
        Return a tuple of dictionaries where element 0 corresponds to the card
        with the maximum gain and element 1 corresponds to the card with the
        maximum loss. The schema for each dictionary is {"cid": [int], "*": float,
        "purchase_price": float} where * is "gain" or "loss", respectively.
    """
    # Connect to local SQL database
    connection: sqlite3.Connection = sqlite3.connect(database_path)
    cursor: sqlite3.Cursor = connection.cursor()

    # Get list of all cards
    sql_command = "select * from purchase_price"
    card_ids: list[tuple[int, float]] = cursor.execute(sql_command).fetchall()

    gain = {"cid": [], "gain": 0.0, "purchase_price": 0.0}
    loss = {"cid": [], "loss": 0.0, "purchase_price": 0.0}

    for _, (cid, purchase_price) in enumerate(card_ids):
        sql_command = f"""select market_value - lag(market_value) over (order by timestamp) 
                          from card_{cid} order by timestamp desc limit 1"""
        result: Optional[float] = cursor.execute(sql_command).fetchone()[0]

        if result is None:
            continue

        movement = round(result, 2)

        # Record card's gain/loss if it is an extreme
        if movement == gain["gain"]:
            gain["cid"].append(cid)

        if movement > gain["gain"]:
            gain["cid"].clear()
            gain["cid"].append(cid)
            gain["gain"] = movement
            gain["purchase_price"] = purchase_price

        if movement == loss["loss"]:
            loss["cid"].append(cid)

        if movement < loss["loss"]:
            loss["cid"].clear()
            loss["cid"].append(cid)
            loss["loss"] = movement
            loss["purchase_price"] = purchase_price

    connection.close()

    return gain, loss

collection_stats(database_path)

Calculates the market value and gain/loss of the entire collection of cards at the timestamps the prices were recorded in the local SQL database.

Parameters:

Name Type Description Default
database_path Path

Path to the local SQL collection database.

 required

Returns:

Type Description
DataFrame

Returns dataframe with schema {"timestamp": pl.String, "market_value": pl.Float64, "profit": pl.Float64}.
Source code in src/mtg_pynance/analysis.py 
 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
def collection_stats(database_path: Path) -> pl.DataFrame:
    """
    Calculates the market value and gain/loss of the entire collection of cards
    at the timestamps the prices were recorded in the local SQL database.

    Parameters
    ----------
    database_path: Path
        Path to the local SQL collection database.

    Returns
    -------
    pl.DataFrame
        Returns dataframe with schema {"timestamp": pl.String,
        "market_value": pl.Float64, "profit": pl.Float64}.
    """
    # Connect to local SQL database
    connection: sqlite3.Connection = sqlite3.connect(database_path)
    cursor: sqlite3.Cursor = connection.cursor()

    # Get list of all cards
    sql_command = "select * from purchase_price"
    card_ids: list[tuple[int, float]] = cursor.execute(sql_command).fetchall()
    collection_stats: list[list[str, float, float]] = []

    # Record stats for each card in the local database
    for _, (cid, purchase_price) in enumerate(card_ids):
        sql_command = f"select * from card_{cid}"
        result: list[tuple[str, float]] = cursor.execute(sql_command).fetchall()
        card_stats = [
            [timestamp, value, value - purchase_price] for timestamp, value in result
        ]
        collection_stats.extend(card_stats)

    connection.close()

    # Convert statistics from list of lists to dataframe
    df = pl.DataFrame(
        collection_stats,
        schema={
            "timestamp": pl.String,
            "market_value": pl.Float64,
            "gain/loss": pl.Float64,
        },
        orient="row",
    )

    # Add together cards that have the same timestamp
    df = df.group_by("timestamp").agg(
        pl.col("market_value").sum(), pl.col("gain/loss").sum()
    )

    # TODO ensure that this will always sort correctly because the timestampsa are strings
    # Sort dataframe from earliest to latest timestamps
    df = df.sort("timestamp")

    return df

delete_card(database_path, cid)

Deletes card from local SQL database. This includes its price table and purchase price information.

Parameters:

Name Type Description Default
database_path Path

Path to the local SQL collection database.

 required
cid int

cid of card in collection file.

 required
Source code in src/mtg_pynance/analysis.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
def delete_card(database_path: Path, cid: int):
    """
    Deletes card from local SQL database. This includes its
    price table and purchase price information.

    Parameters
    ----------
    database_path: Path
        Path to the local SQL collection database.
    cid: int
        cid of card in collection file.
    """
    # Connect to local SQL database
    connection: sqlite3.Connection = sqlite3.connect(database_path)
    cursor: sqlite3.Cursor = connection.cursor()

    # Delete card's price table from database
    sql_command = f"drop table if exists card_{cid}"
    cursor.execute(sql_command)

    # Delete card's purchase price
    sql_command = f"delete from purchase_price where cid = {cid}"
    cursor.execute(sql_command)

    connection.commit()
    connection.close()