Position Synchronizer Development

1. The Principle of Reconciliation

A production-grade trading system cannot blindly trust its own internal state. It must constantly verify that its view of the world matches the broker's "ground truth." This verification process is called reconciliation.

The goal of reconciliation is to ensure that any discrepancies between the system's internal state (what we think we own) and the external state at the trading venue (what we actually own) are detected and corrected.

Reconciliation typically happens in two phases:

1. Startup Reconciliation: A full, comprehensive check when the application first launches. It queries the broker for all open positions and orders to build an accurate initial state before any new trading activity begins.
2. Continuous Reconciliation: Ongoing checks performed while the system is running. Our event-driven processing of fills is one form of this. A more robust implementation might also perform periodic full checks as a safety net.

Today, we will focus primarily on designing and implementing the logic for startup reconciliation.

2. Fetching the Broker's State

Before we can reconcile, we need to get the broker's version of our portfolio. With the native ibapi, this is an asynchronous request/response process, much like placing an order.

1. You call a request function (in this case, reqPositions()).
2. The TWS/Gateway responds by invoking callback methods on your EWrapper instance. The key callbacks for positions are:
   • position(account, contract, pos, avgCost): Called for each position held in the account.
   • positionEnd(): Called once all position data has been sent.

Let's add the necessary logic to our ExecutionHandler.

# In your ExecutionHandler class

    # ... other methods ...

    def __init__(self, event_queue: queue.Queue):
        # ... (previous initializations)
        self.internal_positions = {} # Key: conId, Value: {'contract': Contract, 'position': float, 'avg_cost': float}

    # --- New methods for position fetching ---

    def reqPositions(self):
        """Request all account positions from the broker."""
        print("Requesting account positions...")
        super().reqPositions()

    def position(self, account: str, contract: Contract, pos: float, avgCost: float):
        """Callback receiving position data."""
        super().position(account, contract, pos, avgCost)
        print(f"Position. Account: {account}, Symbol: {contract.symbol}, "
              f"SecType: {contract.secType}, ConID: {contract.conId}, "
              f"Position: {pos}, Avg cost: {avgCost}")
        
        # Store the received position data
        self.internal_positions[contract.conId] = {
            "contract": contract,
            "position": pos,
            "avg_cost": avgCost
        }

    def positionEnd(self):
        """Callback indicating all position data has been received."""
        super().positionEnd()
        print("Position data feed complete.")
        # We can now consider the self.internal_positions dictionary to be fully populated
        # In a real system, you might put a 'PositionsReady' event on the queue here.

The ib_async library, a popular alternative to the native ibapi, provides a simplified, synchronous-style way to do this. While we are using ibapi, looking at an ib_async example clarifies the goal: a simple function call that returns a list of positions.

Crucial Setup Note: For this to work correctly at startup, you must have the correct setting enabled in TWS or IB Gateway.

In TWS/Gateway, go to Configure → API → Settings and ensure "Download open orders on connection" is checked. This tells the broker to send all existing position and order data as soon as our client connects.

3. The Reconciliation Logic: Compare and Correct

Now for the core of the synchronizer. We have two datasets:

1. Internal Portfolio: The state our application has built by processing OrderFillEvents. Let's assume we have a PortfolioManager class that holds this state.
2. Broker Portfolio: The state we just fetched using reqPositions() and stored in our ExecutionHandler.

The reconciliation process involves systematically comparing these two and making corrections. The broker's portfolio is always considered the source of truth.

Let's walk through the logic using the excellent scenarios outlined in the NautilusTrader documentation as a guide.

Here is the pseudocode for a PositionSynchronizer class:

class PositionSynchronizer:
    def __init__(self, internal_portfolio, broker_portfolio):
        # internal_portfolio: Our system's view (e.g., from a PortfolioManager class)
        # broker_portfolio: The view fetched from the broker (e.g., the dictionary we built)
        self.internal_portfolio = internal_portfolio # Format: {conId: position_object}
        self.broker_portfolio = broker_portfolio   # Format: {conId: position_dict}

    def reconcile(self):
        print("--- Starting Position Reconciliation ---")
        internal_conids = set(self.internal_portfolio.keys())
        broker_conids = set(self.broker_portfolio.keys())

        # Scenario 1: Positions that exist in both systems.
        # Check for quantity mismatches.
        shared_conids = internal_conids.intersection(broker_conids)
        for conId in shared_conids:
            internal_pos = self.internal_portfolio[conId].quantity
            broker_pos = self.broker_portfolio[conId]['position']
            if internal_pos != broker_pos:
                print(f"MISMATCH for ConID {conId}: Internal={internal_pos}, Broker={broker_pos}. Correcting internal state.")
                # The broker is truth. Update our internal state.
                self.internal_portfolio[conId].quantity = broker_pos
                self.internal_portfolio[conId].avg_cost = self.broker_portfolio[conId]['avg_cost']

        # Scenario 2: Positions that exist at the broker but not in our system.
        # This means we missed the fill entirely (e.g., system was offline).
        missing_in_internal = broker_conids - internal_conids
        for conId in missing_in_internal:
            broker_pos_data = self.broker_portfolio[conId]
            print(f"MISSING internally for ConID {conId}: Broker has position {broker_pos_data['position']}. Adding to internal state.")
            # Create a new position object in our internal portfolio.
            # self.internal_portfolio[conId] = Position(contract=broker_pos_data['contract'], ...)

        # Scenario 3: Positions that exist in our system but not at the broker.
        # This means we missed a closing fill. This is a "phantom" position.
        missing_in_broker = internal_conids - broker_conids
        for conId in missing_in_broker:
            print(f"PHANTOM position for ConID {conId}: Internal state shows a position, but broker does not. Removing from internal state.")
            # Remove the position from our internal portfolio.
            del self.internal_portfolio[conId]
            
        print("--- Reconciliation Complete ---")

4. Integration into the Framework

The final step is to integrate this process into our application's startup sequence. The main control loop of your framework (which we will build in the next module) should perform these steps in order:

1. Initialize and connect the ExecutionHandler. Wait for the nextValidId callback to confirm the connection is live.
2. Call execution_handler.reqPositions() to request the broker's state.
3. Wait for the positionEnd callback. This can be managed with threading events or by checking a flag.
4. Once positionEnd is received, instantiate the PositionSynchronizer with the newly fetched broker portfolio and your system's current internal portfolio.
5. Run the reconcile() method.
6. Only after reconciliation is complete should the system start its main event loop to process market data and generate new orders.

This ensures you always start from a known, correct state, making your system far more resilient to restarts and intermittent connection issues.

Conclusion

In this lesson, we designed the logic for a critical component of any serious trading system: the position synchronizer. We've established that blind faith in an application's internal state is fragile, and regular reconciliation against the broker's records is mandatory for a "foolproof" system.

Key Takeaways:

• Reconciliation is the process of aligning your system's internal state with the broker's "ground truth."
• The process begins at startup by fetching all current positions from the broker using the reqPositions() call and the corresponding position and positionEnd callbacks.
• The core logic involves a three-way comparison: finding shared positions (and checking for quantity mismatches), finding positions the broker has but you don't (adding them), and finding positions you have but the broker doesn't (removing them).
• The broker's record is always treated as the definitive source of truth.

Preview of the Next Lesson:
We have now designed the major components of our framework: a data model, an execution handler that can place orders and process status updates, and a synchronizer to ensure our position state is accurate. The final piece of the puzzle is to orchestrate all these parts. In the next lesson, we will implement the main event loop to drive the flow of data, strategy evaluation, and execution, tying everything together into a cohesive, running application.