A Blueprint for Automating Price Repairs

The Infrastructure: Data Mapping

To support this repair process, data flows across two core target tables containing raw positions and reference data.

1. Position Storage (geneva_rsl_positions)

This table holds the raw, daily position data extracted from Geneva.

• FTP location:  geneva_rsl_positions/geneva_rsl_positions_YYYYMMDD.csv 
• The Problem: Historically, roughly 10% of these rows exhibit "broken" positions where vital pricing fields—specifically Market_Price_Local
  and Market_Price_Principal—arrive as zero, null, or blank.

2. Security Master Reference (bloomberg_refindex)

This table acts as our static security universe and yellow-key reference index. It does not store historical prices; rather, it maps internal identifiers to the corresponding Bloomberg pricing keys, which we will use to request bloomberg pricing via PX_LAST. This can be pulled dynamically to only request missing securities. To accomplish this task, we create a view to select only RSL positions with null prices for the current day,  and download the results to a csv 

• Download to: bloomberg_security_list/refindex_equities.csv

Isolating the Breaks with SQL

Rather than scanning entire datasets programmatically, we can isolate broken positions at the database level. The view below filters out healthy data, surfacing only rows with missing prices, and maps to a bloomberg security list.

To reliably execute this process, the pipeline is divided into distinct, decoupled workflows. Decoupling ensures that an upstream extraction failure won't inadvertently trigger an incomplete pricing pull.

[ Phase 1: Ingestion ]
   Geneva RSL Extract ──► On Success ──► SQL Bulk Load to Table
                        └──► On Failure ──► Email Alert

[ Phase 2: Price Fetch ]
   Evaluate 'v_rsl_missing_prices' ──► Query Bloomberg API ──► Stage EOD Prices

[ Phase 3: Repair & Load ]
   Download Conditional View ──► Execute Geneva Loader Job

Phase 1: Position Ingestion & Validation

The first workflow runs an RSL extract job from Geneva to pull down current positions straight to the file store.

• Branching Logic: If the file generation fails, an immediate email alert notification triggers, stopping the pipeline.
• Bulk Load: If successful, a SQL Bulk Load process executes, appending the data to our geneva_rsl_positions staging table.

Phase 2: Targeted Price Fetching

Once the positions are staged, the next workflow inspects the v_rsl_missing_prices view. If records exist, a second workflow initiates to pull missing End of Day (EOD) market prices. This script extracts the targeted subset of BBG_KEY fields and calls the Bloomberg API, staging the fresh prices back to the file store.

Phase 3: Conditional Repair & Geneva Ingestion

With both datasets localized, the final phase involves a clean separation between data preparation (the database layer) and data ingestion (the automation layer).

1. The Consolidation View (v_repaired_positions)

Instead of directly modifying the raw ingestion table, we build a dynamic consolidation layer. This view scans the original positions and evaluates the price fields. If the original Geneva price is present and valid, it stands. If it is null, zero, or blank, the view dynamically swaps in the newly fetched Bloomberg price.

SQL

CREATE VIEW v_repaired_positions AS
SELECT 
    pos.As_Of_Date,
    pos.Instrument_ID,
    pos.BBG_Ticker,
    pos.Portfolio_ID,
    pos.Quantity,
    pos.Average_Cost,
    -- Conditionally repair Local Price
    CASE 
        WHEN pos.Market_Price_Local IS NULL
        THEN bbg.EOD_Price_Local
        ELSE pos.Market_Price_Local
    END AS Market_Price_Local,
    -- Conditionally repair Principal Price
    CASE 
        WHEN pos.Market_Price_Principal IS NULL
        THEN bbg.EOD_Price_Principal
        ELSE pos.Market_Price_Principal
    END AS Market_Price_Principal
FROM geneva_rsl_positions pos
LEFT JOIN bloomberg_staged_prices bbg 
    ON pos.bbg_price_key = bbg.bbg_price_key 
   AND pos.As_Of_Date = bbg.As_Of_Date;

2. The Extraction & Loading Workflow

Data pipelines in finance are only as strong as their weakest dependency. A common point of failure occurs during end-of-day operations when upstream position reports drop with missing or zeroed-out market prices. Left unaddressed, these broken positions stall downstream reconciliation, risk reporting, and accounting systems.

To close the loop, I've created a workflow using the CoSet Workflow AI tool, outlining what i want my workflow to do, and following the steps provided by the AI agent, to outline a workflow and fill in the details.

Once the database layer handles the conditional logic, a final decoupled workflow operationalizes the data movement:

• Step 1 (Data Export): A workflow task queries v_repaired_positions specifically filtered for the current run date, exporting the clean, fully-priced records into a flat file on the local file store.
• Step 2 (Geneva Ingestion): Upon a successful export, a Geneva loader job is automatically triggered. This job parses the repaired file and pushes the corrected records straight back into Geneva, completing the automated loop.f

This article walks through a clean blueprint for isolating missing market data, fetching secondary market pricing from Bloomberg, and conditionally repairing records before pushing them back into Geneva.

Putting this all together

Using the steps above, I have an end to end process that does the following at the end of every business day:

• Extracts positions data from my accounting system
• Requests missing prices from Bloomberg
• Transforms my data to fill Bloomberg pricing when my accounting system was missing data
• Pushes updated prices to repair prices not only in CoSet, but also in Geneva

All of the above happens on schedule, with no code and no manual intervention.