You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
64 lines
2.8 KiB
Python
64 lines
2.8 KiB
Python
"""add Form 4 derived columns (is_ceo, is_cfo, is_c_suite, purchase_pct_of_holding, owner_relationship)
|
|
|
|
Revision ID: l3d4e5f6g7h8
|
|
Revises: k2c3d4e5f6g7
|
|
Create Date: 2026-04-22
|
|
|
|
"""
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
|
|
revision = "l3d4e5f6g7h8"
|
|
down_revision = "k2c3d4e5f6g7"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
|
|
def upgrade() -> None:
|
|
conn = op.get_bind()
|
|
|
|
# Add new columns
|
|
op.add_column("insider_transactions", sa.Column("owner_relationship", sa.Text(), nullable=True))
|
|
op.add_column("insider_transactions", sa.Column("is_ceo", sa.Boolean(), nullable=False, server_default="false"))
|
|
op.add_column("insider_transactions", sa.Column("is_cfo", sa.Boolean(), nullable=False, server_default="false"))
|
|
op.add_column("insider_transactions", sa.Column("is_c_suite", sa.Boolean(), nullable=False, server_default="false"))
|
|
op.add_column("insider_transactions", sa.Column("purchase_pct_of_holding", sa.Float(), nullable=True))
|
|
|
|
# Widen transaction_code from varchar(5) to varchar(10)
|
|
op.alter_column("insider_transactions", "transaction_code",
|
|
existing_type=sa.String(5), type_=sa.String(10))
|
|
|
|
# New indexes for PIT / by-date queries
|
|
op.create_index("idx_insider_filing_ticker", "insider_transactions", ["filing_date", "ticker"])
|
|
|
|
# Backfill c-suite flags from existing officer_title data
|
|
conn.execute(sa.text("""
|
|
UPDATE insider_transactions
|
|
SET
|
|
is_ceo = (officer_title ~* '\\mCEO\\M|Chief\\s+Executive\\s+Officer'),
|
|
is_cfo = (officer_title ~* '\\mCFO\\M|Chief\\s+Financial\\s+Officer|Principal\\s+Financial\\s+Officer'),
|
|
is_c_suite = (officer_title ~* '\\mCEO\\M|Chief\\s+Executive\\s+Officer|\\mCFO\\M|Chief\\s+Financial\\s+Officer|Principal\\s+Financial\\s+Officer|\\mCOO\\M|\\mCTO\\M|\\mCIO\\M|\\mCLO\\M|\\mCMO\\M|\\mPresident\\M|\\mChair(man|person|woman)?\\M|Chief\\s+\\w+\\s+Officer')
|
|
WHERE officer_title IS NOT NULL
|
|
"""))
|
|
|
|
# Backfill purchase_pct_of_holding for existing buy transactions
|
|
conn.execute(sa.text("""
|
|
UPDATE insider_transactions
|
|
SET purchase_pct_of_holding = ABS(shares) / NULLIF(shares_owned_after, 0)
|
|
WHERE transaction_code IN ('P', 'A')
|
|
AND shares > 0
|
|
AND shares_owned_after IS NOT NULL
|
|
AND shares_owned_after > 0
|
|
"""))
|
|
|
|
|
|
def downgrade() -> None:
|
|
op.drop_index("idx_insider_filing_ticker", table_name="insider_transactions")
|
|
op.drop_column("insider_transactions", "purchase_pct_of_holding")
|
|
op.drop_column("insider_transactions", "is_c_suite")
|
|
op.drop_column("insider_transactions", "is_cfo")
|
|
op.drop_column("insider_transactions", "is_ceo")
|
|
op.drop_column("insider_transactions", "owner_relationship")
|
|
op.alter_column("insider_transactions", "transaction_code",
|
|
existing_type=sa.String(10), type_=sa.String(5))
|