@
R3 Slice 4: IBKR-Marktdaten auf EF; Dapper vollstaendig entfernt - 3 IBKR-Entities (IBKRInstrument/IBKRMarketBar/IBKRExternalIdentifier) im CoreDbContext (core_ibkr_instruments/_market_data/_external_identifiers); Migration AddIbkr - IBKRMarketDataRepository von Dapper auf EF (IDbContextFactory<CoreDbContext>); Cross-Modul-Query (ct_trade) via Database.SqlQueryRaw - DatabaseService, IBKRMigrations, Dapper-Package entfernt; Laufzeit-Migrationen komplett weg (Schema extern via dotnet ef database update) - Tests: +4 IBKRMarketDataRepository (EF-InMemory) -> 43/43 gruen; Build + smoke-ui + App-Start ok R3 abgeschlossen: Persistenz vollstaendig auf EF Core. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> @
This commit is contained in:
Richard
@@ -1,152 +1,154 @@
|
||||
using Dapper;
|
||||
using IBKRTrader.Core.Database;
|
||||
using IBKRTrader.Core.Logging;
|
||||
using IBKRTrader.Core.Persistence.Ef;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
|
||||
namespace IBKRTrader.Core.IBKR;
|
||||
|
||||
/// <summary>
|
||||
/// Datenzugriffsschicht für die core_ibkr_xxx-Tabellen.
|
||||
/// Alle IBKR-Marktdaten-Operationen laufen über diese Klasse.
|
||||
/// Datenzugriffsschicht für die core_ibkr_-Tabellen (EF Core).
|
||||
/// </summary>
|
||||
public class IBKRMarketDataRepository
|
||||
{
|
||||
private readonly DatabaseService _db;
|
||||
private readonly LoggingService _logger;
|
||||
private readonly IDbContextFactory<CoreDbContext> _dbf;
|
||||
private readonly LoggingService _logger;
|
||||
|
||||
public IBKRMarketDataRepository(DatabaseService db, LoggingService logger)
|
||||
public IBKRMarketDataRepository(IDbContextFactory<CoreDbContext> dbf, LoggingService logger)
|
||||
{
|
||||
_db = db;
|
||||
_dbf = dbf;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
// ─── Instruments ─────────────────────────────────────────────────────────
|
||||
|
||||
/// <summary>
|
||||
/// Legt ein neues Instrument an oder aktualisiert ein bestehendes (UPSERT via conid).
|
||||
/// Gibt die Instrument-ID zurück.
|
||||
/// </summary>
|
||||
/// <summary>Upsert eines Instruments (per ConID). Gibt die Instrument-ID zurück.</summary>
|
||||
public async Task<long> UpsertInstrumentAsync(IBKRInstrument instr)
|
||||
{
|
||||
const string sql = @"
|
||||
INSERT INTO `core_ibkr_instruments`
|
||||
(`ibkr_conid`, `symbol`, `sec_type`, `exchange`, `primary_exchange`,
|
||||
`currency`, `company_name`, `isin`, `sector`, `industry`,
|
||||
`description`, `active`, `last_fetched`)
|
||||
VALUES
|
||||
(@IbkrConid, @Symbol, @SecType, @Exchange, @PrimaryExchange,
|
||||
@Currency, @CompanyName, @Isin, @Sector, @Industry,
|
||||
@Description, @Active, @LastFetched)
|
||||
ON DUPLICATE KEY UPDATE
|
||||
`symbol` = VALUES(`symbol`),
|
||||
`sec_type` = VALUES(`sec_type`),
|
||||
`exchange` = VALUES(`exchange`),
|
||||
`primary_exchange` = VALUES(`primary_exchange`),
|
||||
`currency` = VALUES(`currency`),
|
||||
`company_name` = VALUES(`company_name`),
|
||||
`isin` = VALUES(`isin`),
|
||||
`sector` = VALUES(`sector`),
|
||||
`industry` = VALUES(`industry`),
|
||||
`description` = VALUES(`description`),
|
||||
`last_fetched` = VALUES(`last_fetched`);
|
||||
SELECT `id` FROM `core_ibkr_instruments` WHERE `ibkr_conid` = @IbkrConid;";
|
||||
await using var db = await _dbf.CreateDbContextAsync();
|
||||
var existing = await db.Instruments.FirstOrDefaultAsync(i => i.IbkrConid == instr.IbkrConid);
|
||||
if (existing is null)
|
||||
{
|
||||
instr.CreatedAt = DateTime.UtcNow;
|
||||
instr.UpdatedAt = DateTime.UtcNow;
|
||||
db.Instruments.Add(instr);
|
||||
await db.SaveChangesAsync();
|
||||
return instr.Id;
|
||||
}
|
||||
|
||||
await using var conn = _db.CreateConnection();
|
||||
return await conn.ExecuteScalarAsync<long>(sql, instr);
|
||||
existing.Symbol = instr.Symbol;
|
||||
existing.SecType = instr.SecType;
|
||||
existing.Exchange = instr.Exchange;
|
||||
existing.PrimaryExchange = instr.PrimaryExchange;
|
||||
existing.Currency = instr.Currency;
|
||||
existing.CompanyName = instr.CompanyName;
|
||||
existing.Isin = instr.Isin;
|
||||
existing.Sector = instr.Sector;
|
||||
existing.Industry = instr.Industry;
|
||||
existing.Description = instr.Description;
|
||||
existing.LastFetched = instr.LastFetched;
|
||||
existing.UpdatedAt = DateTime.UtcNow;
|
||||
await db.SaveChangesAsync();
|
||||
return existing.Id;
|
||||
}
|
||||
|
||||
/// <summary>Gibt alle aktiven Instrumente zurück.</summary>
|
||||
public Task<IEnumerable<IBKRInstrument>> GetAllActiveInstrumentsAsync()
|
||||
=> _db.QueryAsync<IBKRInstrument>(
|
||||
"SELECT * FROM `core_ibkr_instruments` WHERE `active` = 1 ORDER BY `symbol`");
|
||||
public async Task<IEnumerable<IBKRInstrument>> GetAllActiveInstrumentsAsync()
|
||||
{
|
||||
await using var db = await _dbf.CreateDbContextAsync();
|
||||
return await db.Instruments.Where(i => i.Active).OrderBy(i => i.Symbol).ToListAsync();
|
||||
}
|
||||
|
||||
/// <summary>Findet ein Instrument anhand seiner IBKR ConID.</summary>
|
||||
public Task<IBKRInstrument?> GetInstrumentByConidAsync(long conid)
|
||||
=> _db.QueryFirstOrDefaultAsync<IBKRInstrument>(
|
||||
"SELECT * FROM `core_ibkr_instruments` WHERE `ibkr_conid` = @conid",
|
||||
new { conid });
|
||||
public async Task<IBKRInstrument?> GetInstrumentByConidAsync(long conid)
|
||||
{
|
||||
await using var db = await _dbf.CreateDbContextAsync();
|
||||
return await db.Instruments.FirstOrDefaultAsync(i => i.IbkrConid == conid);
|
||||
}
|
||||
|
||||
/// <summary>Gibt die Anzahl aktiver Instrumente zurück.</summary>
|
||||
public Task<int> GetActiveInstrumentCountAsync()
|
||||
=> _db.ExecuteScalarAsync<int>(
|
||||
"SELECT COUNT(*) FROM `core_ibkr_instruments` WHERE `active` = 1");
|
||||
public async Task<int> GetActiveInstrumentCountAsync()
|
||||
{
|
||||
await using var db = await _dbf.CreateDbContextAsync();
|
||||
return await db.Instruments.CountAsync(i => i.Active);
|
||||
}
|
||||
|
||||
// ─── Market Data ─────────────────────────────────────────────────────────
|
||||
|
||||
/// <summary>
|
||||
/// Fügt Marktdaten-Balken via UPSERT ein (ON DUPLICATE KEY UPDATE).
|
||||
/// </summary>
|
||||
/// <summary>Upsert einer Menge OHLCV-Balken (per PK InstrumentId+BarSize+Timestamp).</summary>
|
||||
public async Task UpsertMarketDataBatchAsync(IEnumerable<IBKRMarketBar> bars)
|
||||
{
|
||||
const string sql = @"
|
||||
INSERT INTO `core_ibkr_market_data`
|
||||
(`instrument_id`, `bar_size`, `timestamp`,
|
||||
`open`, `high`, `low`, `close`, `volume`, `wap`, `bar_count`)
|
||||
VALUES
|
||||
(@InstrumentId, @BarSize, @Timestamp,
|
||||
@Open, @High, @Low, @Close, @Volume, @Wap, @BarCount)
|
||||
ON DUPLICATE KEY UPDATE
|
||||
`open` = VALUES(`open`),
|
||||
`high` = VALUES(`high`),
|
||||
`low` = VALUES(`low`),
|
||||
`close` = VALUES(`close`),
|
||||
`volume` = VALUES(`volume`),
|
||||
`wap` = VALUES(`wap`),
|
||||
`bar_count` = VALUES(`bar_count`)";
|
||||
|
||||
await using var conn = _db.CreateConnection();
|
||||
await conn.OpenAsync();
|
||||
await conn.ExecuteAsync(sql, bars);
|
||||
await using var db = await _dbf.CreateDbContextAsync();
|
||||
foreach (var bar in bars)
|
||||
{
|
||||
var existing = await db.MarketBars.FindAsync(bar.InstrumentId, bar.BarSize, bar.Timestamp);
|
||||
if (existing is null)
|
||||
{
|
||||
db.MarketBars.Add(bar);
|
||||
}
|
||||
else
|
||||
{
|
||||
existing.Open = bar.Open;
|
||||
existing.High = bar.High;
|
||||
existing.Low = bar.Low;
|
||||
existing.Close = bar.Close;
|
||||
existing.Volume = bar.Volume;
|
||||
existing.Wap = bar.Wap;
|
||||
existing.BarCount = bar.BarCount;
|
||||
}
|
||||
}
|
||||
await db.SaveChangesAsync();
|
||||
}
|
||||
|
||||
/// <summary>Gibt den neuesten Timestamp für ein Instrument zurück.</summary>
|
||||
public Task<DateTime?> GetLatestBarTimestampAsync(long instrumentId, string barSize = "daily")
|
||||
=> _db.QueryFirstOrDefaultAsync<DateTime?>(
|
||||
@"SELECT MAX(`timestamp`) FROM `core_ibkr_market_data`
|
||||
WHERE `instrument_id` = @instrumentId AND `bar_size` = @barSize",
|
||||
new { instrumentId, barSize });
|
||||
public async Task<DateTime?> GetLatestBarTimestampAsync(long instrumentId, string barSize = "daily")
|
||||
{
|
||||
await using var db = await _dbf.CreateDbContextAsync();
|
||||
var bars = db.MarketBars.Where(m => m.InstrumentId == instrumentId && m.BarSize == barSize);
|
||||
if (!await bars.AnyAsync()) return null;
|
||||
return await bars.MaxAsync(m => m.Timestamp);
|
||||
}
|
||||
|
||||
/// <summary>Gibt die Anzahl Bars für ein Instrument zurück.</summary>
|
||||
public Task<int> GetBarCountAsync(long instrumentId, string barSize = "daily")
|
||||
=> _db.ExecuteScalarAsync<int>(
|
||||
@"SELECT COUNT(*) FROM `core_ibkr_market_data`
|
||||
WHERE `instrument_id` = @instrumentId AND `bar_size` = @barSize",
|
||||
new { instrumentId, barSize });
|
||||
public async Task<int> GetBarCountAsync(long instrumentId, string barSize = "daily")
|
||||
{
|
||||
await using var db = await _dbf.CreateDbContextAsync();
|
||||
return await db.MarketBars.CountAsync(m => m.InstrumentId == instrumentId && m.BarSize == barSize);
|
||||
}
|
||||
|
||||
// ─── External Identifiers ────────────────────────────────────────────────
|
||||
|
||||
/// <summary>
|
||||
/// Erstellt oder ignoriert ein External-Identifier-Mapping (IGNORE bei Duplikat).
|
||||
/// </summary>
|
||||
public Task UpsertExternalIdentifierAsync(long instrumentId, string source, string ticker)
|
||||
=> _db.ExecuteAsync(@"
|
||||
INSERT IGNORE INTO `core_ibkr_external_identifiers`
|
||||
(`instrument_id`, `source`, `ticker`)
|
||||
VALUES (@instrumentId, @source, @ticker)",
|
||||
new { instrumentId, source, ticker });
|
||||
/// <summary>Legt ein Ticker-Mapping an (ignoriert Duplikate).</summary>
|
||||
public async Task UpsertExternalIdentifierAsync(long instrumentId, string source, string ticker)
|
||||
{
|
||||
await using var db = await _dbf.CreateDbContextAsync();
|
||||
var exists = await db.ExternalIds.AnyAsync(
|
||||
e => e.InstrumentId == instrumentId && e.Source == source && e.Ticker == ticker);
|
||||
if (exists) return;
|
||||
|
||||
db.ExternalIds.Add(new IBKRExternalIdentifier { InstrumentId = instrumentId, Source = source, Ticker = ticker });
|
||||
await db.SaveChangesAsync();
|
||||
}
|
||||
|
||||
public async Task<IBKRInstrument?> FindInstrumentByExternalTickerAsync(string source, string ticker)
|
||||
{
|
||||
await using var db = await _dbf.CreateDbContextAsync();
|
||||
var query =
|
||||
from e in db.ExternalIds
|
||||
join i in db.Instruments on e.InstrumentId equals i.Id
|
||||
where e.Source == source && e.Ticker == ticker
|
||||
select i;
|
||||
return await query.FirstOrDefaultAsync();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Findet ein Instrument anhand eines externen Tickers (z.B. aus ct_trade).
|
||||
/// Findet alle Ticker aus ct_trade (Modul-Tabelle) ohne IBKR-Mapping. Cross-Tabellen-Query via
|
||||
/// Raw-SQL, damit der Core den Modul-DbContext nicht referenzieren muss.
|
||||
/// </summary>
|
||||
public Task<IBKRInstrument?> FindInstrumentByExternalTickerAsync(string source, string ticker)
|
||||
=> _db.QueryFirstOrDefaultAsync<IBKRInstrument>(@"
|
||||
SELECT i.* FROM `core_ibkr_instruments` i
|
||||
INNER JOIN `core_ibkr_external_identifiers` e ON e.`instrument_id` = i.`id`
|
||||
WHERE e.`source` = @source AND e.`ticker` = @ticker
|
||||
LIMIT 1",
|
||||
new { source, ticker });
|
||||
|
||||
/// <summary>
|
||||
/// Findet alle einzigartigen Ticker aus ct_trade, die noch kein IBKR-Mapping haben.
|
||||
/// </summary>
|
||||
public Task<IEnumerable<string>> GetUnmappedTickersFromCongressTradesAsync()
|
||||
=> _db.QueryAsync<string>(@"
|
||||
SELECT DISTINCT t.`ticker`
|
||||
public async Task<IEnumerable<string>> GetUnmappedTickersFromCongressTradesAsync()
|
||||
{
|
||||
await using var db = await _dbf.CreateDbContextAsync();
|
||||
const string sql = @"
|
||||
SELECT DISTINCT t.`Ticker` AS `Value`
|
||||
FROM `ct_trade` t
|
||||
WHERE t.`ticker` IS NOT NULL
|
||||
AND t.`ticker` != ''
|
||||
WHERE t.`Ticker` IS NOT NULL AND t.`Ticker` <> ''
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM `core_ibkr_external_identifiers` e
|
||||
WHERE e.`source` = 'capitoltrades' AND e.`ticker` = t.`ticker`
|
||||
)
|
||||
ORDER BY t.`ticker`");
|
||||
WHERE e.`Source` = 'capitoltrades' AND e.`Ticker` = t.`Ticker`)
|
||||
ORDER BY t.`Ticker`";
|
||||
return await db.Database.SqlQueryRaw<string>(sql).ToListAsync();
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user