# models.py from tortoise import fields, models from typing import Optional from datetime import datetime from tortoise.contrib.pydantic.creator import pydantic_model_creator, pydantic_queryset_creator from tortoise.queryset import QuerySet class Portfolio(models.Model): id = fields.IntField(pk=True) user = fields.ForeignKeyField("models.User", related_name="portfolios") name = fields.CharField(max_length=100) description = fields.TextField(null=True) is_active = fields.BooleanField(default=True) created_at = fields.DatetimeField(auto_now_add=True) updated_at = fields.DatetimeField(auto_now=True) async def to_dict(self): if type(self) == models.Model: parser = pydantic_model_creator(Portfolio) return await parser.from_tortoise_orm(self) if type(self) == QuerySet: parser = pydantic_queryset_creator(Portfolio) return await parser.from_queryset(self) class Meta: table = "portfolios" unique_together = ("user", "name") # User can't have duplicate portfolio names class PortfolioStock(models.Model): id = fields.IntField(pk=True) portfolio = fields.ForeignKeyField("models.Portfolio", related_name="stocks") stock = fields.ForeignKeyField("models.Stock", related_name="portfolio_holdings") quantity = fields.IntField() purchase_price = fields.DecimalField(max_digits=15, decimal_places=2) purchase_date = fields.DateField() notes = fields.TextField(null=True) created_at = fields.DatetimeField(auto_now_add=True) updated_at = fields.DatetimeField(auto_now=True) async def to_dict(self): if type(self) == models.Model: parser = pydantic_model_creator(PortfolioStock) return await parser.from_tortoise_orm(self) if type(self) == QuerySet: parser = pydantic_queryset_creator(PortfolioStock) return await parser.from_queryset(self) class Meta: table = "portfolio_stocks" class PortfolioUTT(models.Model): id = fields.IntField(pk=True) portfolio = fields.ForeignKeyField("models.Portfolio", related_name="utts") utt_fund = fields.ForeignKeyField("models.UTTFund", related_name="portfolio_holdings") units_held = fields.DecimalField(max_digits=15, decimal_places=4) purchase_price = fields.DecimalField(max_digits=15, decimal_places=2) purchase_date = fields.DateField() notes = fields.TextField(null=True) created_at = fields.DatetimeField(auto_now_add=True) updated_at = fields.DatetimeField(auto_now=True) async def to_dict(self): if type(self) == models.Model: parser = pydantic_model_creator(PortfolioUTT) return await parser.from_tortoise_orm(self) if type(self) == QuerySet: parser = pydantic_queryset_creator(PortfolioUTT) return await parser.from_queryset(self) class Meta: table = "portfolio_utts" class PortfolioBond(models.Model): id = fields.IntField(pk=True) portfolio = fields.ForeignKeyField("models.Portfolio", related_name="bonds") bond = fields.ForeignKeyField("models.Bond", related_name="portfolio_holdings") face_value_held = fields.BigIntField() purchase_price = fields.DecimalField(max_digits=15, decimal_places=2) purchase_date = fields.DateField() notes = fields.TextField(null=True) created_at = fields.DatetimeField(auto_now_add=True) updated_at = fields.DatetimeField(auto_now=True) async def to_dict(self): if type(self) == models.Model: parser = pydantic_model_creator(PortfolioBond) return await parser.from_tortoise_orm(self) if type(self) == QuerySet: parser = pydantic_queryset_creator(PortfolioBond) return await parser.from_queryset(self) class Meta: table = "portfolio_bonds" class PortfolioTransaction(models.Model): """Track all portfolio transactions for audit and reporting""" id = fields.IntField(pk=True) portfolio = fields.ForeignKeyField("models.Portfolio", related_name="transactions") transaction_type = fields.CharField(max_length=20) # BUY, SELL, DIVIDEND, COUPON asset_type = fields.CharField(max_length=10) # STOCK, BOND, UTT asset_id = fields.IntField() # Generic reference to stock/bond/utt ID quantity = fields.DecimalField(max_digits=15, decimal_places=4) price = fields.DecimalField(max_digits=15, decimal_places=2) total_amount = fields.DecimalField(max_digits=15, decimal_places=2) transaction_date = fields.DateField() notes = fields.TextField(null=True) created_at = fields.DatetimeField(auto_now_add=True) @staticmethod async def get_list(data): if type(data) == QuerySet: parser = pydantic_queryset_creator(PortfolioTransaction) return await parser.from_queryset(data) async def to_dict(self): if type(self) == models.Model: parser = pydantic_model_creator(PortfolioTransaction) return await parser.from_tortoise_orm(self) class Meta: table = "portfolio_transactions" class PortfolioCalendar(models.Model): id = fields.IntField(pk=True) portfolio = fields.ForeignKeyField("models.Portfolio", related_name="calendar_events") event_date = fields.DateField() event_type = fields.CharField(max_length=50) # COUPON, DIVIDEND, MATURITY, EARNINGS title = fields.CharField(max_length=200) description = fields.TextField(null=True) asset_type = fields.CharField(max_length=10, null=True) # STOCK, BOND, UTT asset_id = fields.IntField(null=True) estimated_amount = fields.DecimalField(max_digits=15, decimal_places=2, null=True) is_completed = fields.BooleanField(default=False) created_at = fields.DatetimeField(auto_now_add=True) @staticmethod async def get_list(data): if type(data) == QuerySet: parser = pydantic_queryset_creator(PortfolioCalendar) return await parser.from_queryset(data) async def to_dict(self): if type(self) == models.Model: parser = pydantic_model_creator(PortfolioCalendar) return await parser.from_tortoise_orm(self) class Meta: table = "portfolio_calendar" class PortfolioSnapshot(models.Model): """Daily snapshots for performance tracking""" id = fields.IntField(pk=True) portfolio = fields.ForeignKeyField("models.Portfolio", related_name="snapshots") snapshot_date = fields.DatetimeField() total_value = fields.DecimalField(max_digits=20, decimal_places=2) stock_value = fields.DecimalField(max_digits=20, decimal_places=2, default=0) bond_value = fields.DecimalField(max_digits=20, decimal_places=2, default=0) utt_value = fields.DecimalField(max_digits=20, decimal_places=2, default=0) cash_value = fields.DecimalField(max_digits=20, decimal_places=2, default=0) total_cost = fields.DecimalField(max_digits=20, decimal_places=2) unrealized_gain_loss = fields.DecimalField(max_digits=20, decimal_places=2) created_at = fields.DatetimeField(auto_now_add=True) @staticmethod async def get_list(data): if type(data) == QuerySet: parser = pydantic_queryset_creator(PortfolioSnapshot) return await parser.from_queryset(data) async def to_dict(self): if type(self) == models.Model: parser = pydantic_model_creator(PortfolioSnapshot) return await parser.from_tortoise_orm(self) class Meta: table = "portfolio_snapshots" unique_together = ("portfolio", "snapshot_date")